综合实战：订单事件系统的选型与判别

Properties p = new Properties();
p.put("bootstrap.servers", "broker1:9092,broker2:9092,broker3:9092");
p.put("key.serializer",   "org.apache.kafka.common.serialization.StringSerializer");
p.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// —— 持久性四件套里的 producer 端两项（其余两项在 topic/broker 端，见下）——
p.put("acks", "all");              // 等所有 ISR 成员确认，而非 leader 单方面（堵陷阱 1）
p.put("enable.idempotence", "true"); // PID+序列号去重并强制保序；in-flight<=5 仍有序（堵正确性类乱序）
// enable.idempotence=true 会隐式要求 acks=all、retries>0，4.0 起默认即开

KafkaProducer<String, String> producer = new KafkaProducer<>(p);

void publish(OrderEvent e) {
    // 关键：key = orderId（高基数、避免热分区；同单事件同分区 → 分区内有序）
    var record = new ProducerRecord<>("orders", e.getOrderId(), e.toJson());
    producer.send(record, (md, ex) -> {
        if (ex != null) handleSendFailure(e, ex);   // 失败要可观测，别吞异常
    });
}

# 持久性四件套的另外两项 + 不可逆截断的闸门
replication.factor=3                  # 三副本，单 broker 挂不丢分区（堵陷阱 4）
min.insync.replicas=2                 # = RF-1：挂一台仍可写、挂两台拒写（堵陷阱 3 的静默退化）
unclean.leader.election.enable=false  # 禁止落后副本上位截断已提交记录（堵陷阱 2）

# 分区数：按峰值吞吐与消费并行预估，且预留余量——加分区会破坏 key 顺序、且不可减
num.partitions=3

Properties c = new Properties();
c.put("bootstrap.servers", "broker1:9092,broker2:9092,broker3:9092");
c.put("group.id", "inventory");        // 独立消费组：与通知/对账各自维护 offset
c.put("key.deserializer",   "org.apache.kafka.common.serialization.StringDeserializer");
c.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
c.put("enable.auto.commit", "false");  // 关掉定时提交：否则提交的是 poll 返回的、不是处理完的（堵陷阱 5）

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(c);
consumer.subscribe(List.of("orders"));

while (running) {
    var records = consumer.poll(Duration.ofMillis(500));
    for (var r : records) {
        OrderEvent e = OrderEvent.parse(r.value());
        if (e.getType() == PAID) {
            // 去重落在 DB 侧：幂等键 = orderId + 事件类型，唯一约束/upsert 挡住重投
            // 即使这批被重投（at-least-once），DB 的扣减结果与处理一次相同
            inventoryDb.deductIdempotent(e.getOrderId(), e.getSku(), e.getQty());
        }
    }
    consumer.commitSync();   // 整批处理完再提交：崩溃只会重投（可被幂等吸收），不会丢
}