项目实战：订单事件流水线

                         +--------------------------+
 订单服务 ──(事务发送)──> | Kafka                     |
                         |  orders.events (12 分区)  |
                         |  orders.events.dlq        |
                         +--------------------------+
                            |            |         |
                     消费组: inventory  risk      dwh-etl
                            |            |         |
                        库存扣减      风控规则    Streams 聚合
                       (幂等写库)    (可容忍重复)  (exactly-once)
                            |
                        失败重试 3 次仍失败 -> DLQ

# 主事件流: 12 分区 (峰值 3 万 msg/s / 单分区 3 千 ≈ 10, 取 12 留余量)
bin/kafka-topics.sh --bootstrap-server localhost:9092 --create \
  --topic orders.events --partitions 12 --replication-factor 3 \
  --config min.insync.replicas=2 \
  --config retention.ms=604800000
 
# 死信队列: 流量极小, 保留期放长便于排查
bin/kafka-topics.sh --bootstrap-server localhost:9092 --create \
  --topic orders.events.dlq --partitions 3 --replication-factor 3 \
  --config retention.ms=2592000000

{
  "eventId": "evt_8f3ac2",
  "eventType": "ORDER_PAID",
  "orderId": "ord_10091",
  "userId": "u_557",
  "amount": 129.00,
  "occurredAt": "2026-07-30T10:15:30Z",
  "schemaVersion": 1
}

public class OrderEventProducer {
    private final KafkaProducer<String, String> producer;
 
    public OrderEventProducer() {
        Properties p = new Properties();
        p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka1:9092,kafka2:9092");
        p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        p.put(ProducerConfig.ACKS_CONFIG, "all");
        p.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        p.put(ProducerConfig.LINGER_MS_CONFIG, 10);
        p.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
        this.producer = new KafkaProducer<>(p);
    }
 
    // key = orderId: 同一订单事件永远同分区 => 局部有序
    public void publish(OrderEvent e) {
        var record = new ProducerRecord<>("orders.events", e.orderId(), toJson(e));
        producer.send(record, (meta, ex) -> {
            if (ex != null) {
                // 发送失败: 落 outbox 表, 由后台任务重发, 保证最终送达
                outboxDao.save(e);
            }
        });
    }
}

public class InventoryConsumer {
    private static final int MAX_RETRIES = 3;
    private final KafkaConsumer<String, String> consumer;
    private final KafkaProducer<String, String> dlqProducer;
 
    public void run() {
        consumer.subscribe(List.of("orders.events"));
        while (running) {
            var records = consumer.poll(Duration.ofMillis(500));
            for (var r : records) {
                handleWithRetry(r);
            }
            consumer.commitSync();   // 整批处理完(含进 DLQ)才提交
        }
    }
 
    private void handleWithRetry(ConsumerRecord<String, String> r) {
        for (int attempt = 1; ; attempt++) {
            try {
                handle(r);
                return;
            } catch (RetriableException e) {          // 网络/超时类: 重试
                if (attempt >= MAX_RETRIES) { sendToDlq(r, e); return; }
                sleepBackoff(attempt);                 // 1s, 2s, 4s...
            } catch (Exception e) {                    // 解析失败等毒消息: 直接 DLQ
                sendToDlq(r, e);
                return;
            }
        }
    }
 
    private void handle(ConsumerRecord<String, String> r) {
        OrderEvent e = parse(r.value());
        if (!"ORDER_PAID".equals(e.eventType())) return;
        // 幂等核心: processed_events.event_id 唯一索引
        // INSERT INTO processed_events(event_id) VALUES (?)
        //   ON CONFLICT DO NOTHING; 影响行数为 0 => 已处理过, 跳过
        boolean firstTime = dedupDao.tryInsert(e.eventId());
        if (!firstTime) return;
        inventoryDao.deduct(e.orderId());              // 与上一句同一个 DB 事务!
    }
 
    private void sendToDlq(ConsumerRecord<String, String> r, Exception cause) {
        var dlqRecord = new ProducerRecord<>("orders.events.dlq", r.key(), r.value());
        dlqRecord.headers()
            .add("dlq.error", cause.toString().getBytes(StandardCharsets.UTF_8))
            .add("dlq.source.partition", String.valueOf(r.partition()).getBytes())
            .add("dlq.source.offset", String.valueOf(r.offset()).getBytes());
        dlqProducer.send(dlqRecord);
        alert("order event moved to DLQ: " + r.key());  // DLQ 必须有告警!
    }
}

def handle(msg):
    event = json.loads(msg.value())
    if event["eventType"] != "ORDER_PAID":
        return
    with db.transaction():
        # 唯一键幂等: 重复 eventId 直接跳过
        inserted = db.execute(
            "INSERT INTO processed_events(event_id) VALUES (%s) "
            "ON CONFLICT DO NOTHING", (event["eventId"],))
        if inserted == 0:
            return
        db.execute("UPDATE inventory SET stock = stock - 1 "
                   "WHERE order_id = %s", (event["orderId"],))
 
while True:
    msg = consumer.poll(0.5)
    if msg is None or msg.error():
        continue
    try:
        retry(handle, msg, times=3, backoff=1.0)
    except Exception as e:
        send_to_dlq(msg, e)
    consumer.commit(message=msg, asynchronous=False)

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "dwh-order-agg");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka1:9092");
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
 
StreamsBuilder builder = new StreamsBuilder();
builder.stream("orders.events", Consumed.with(Serdes.String(), Serdes.String()))
    .filter((k, v) -> v.contains("\"ORDER_PAID\""))
    .mapValues(v -> parseAmount(v))
    .groupBy((k, amt) -> "gmv")                        // 全局聚合
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofHours(1)))
    .reduce(Double::sum, Materialized.as("gmv-hourly"))
    .toStream()
    .map((win, gmv) -> KeyValue.pair(win.key() + "@" + win.window().start(),
                                     String.valueOf(gmv)))
    .to("dwh.gmv.hourly");
 
new KafkaStreams(builder.build(), props).start();