Kafka 怎么保证消费只消费一次的？

结论：Kafka 支持三种投递语义，但"只消费一次"（Exactly-Once）只在 Kafka 内部闭环（消费—处理—生产—提交位点）时才成立，靠事务 + read_committed 实现；一旦处理过程有外部副作用（写 MySQL、调第三方），就必须靠幂等或事务性输出，不能指望 Kafka 单方面保证。

一、三种语义

At-Most-Once    先提交位点，再处理    ── 崩溃 → 消息丢，绝不重复
At-Least-Once   先处理，再提交位点    ── 崩溃 → 重复，绝不丢（默认推荐）
Exactly-Once    处理与位点在同一事务  ── 需要事务 API + read_committed

二、事务如何做到端到端

Consumer（read_committed）
   │ 1. poll 消息
   ├──────────────────────────────────────────────┐
   │ 2. producer.beginTransaction()               │
   │ 3. 处理 + producer.send(输出 Topic)           │ 同一个事务
   │ 4. producer.sendOffsetsToTransaction(位点)    │
   │ 5. producer.commitTransaction()              │
   └──────────────────────────────────────────────┘
                    │ 提交成功
                    ▼
   下游 read_committed 消费者只能看到已提交事务的数据
   未提交/已回滚的数据被 Aborted Transaction 标记屏蔽（LSO 之前）

关键点：位点提交也放进事务。如果位点单独提交，事务回滚后位点已经前移，消息就丢了；放在同一个事务里，二者要么都生效、要么都不生效。

三、代码实现

Properties p = new Properties();
p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.0.0.31:9092");
p.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-etl-1");   // 必须唯一且稳定
p.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
KafkaProducer<String, String> producer = new KafkaProducer<>(p);
producer.initTransactions();                                     // 只调一次

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(200));
    if (records.isEmpty()) continue;
    producer.beginTransaction();
    try {
        for (ConsumerRecord<String, String> r : records) {
            producer.send(new ProducerRecord<>("order-clean", r.key(), transform(r)));
        }
        producer.sendOffsetsToTransaction(currentOffsets(records), consumer.groupMetadata());
        producer.commitTransaction();
    } catch (Exception e) {
        producer.abortTransaction();        // 回滚，位点不前进
    }
}

# 下游消费者必须用 read_committed，否则会读到未提交数据
isolation.level=read_committed
enable.auto.commit=false

四、有外部副作用时怎么办

Kafka 事务管不到 MySQL。常见三种工程做法：

绝大多数团队选择"至少一次 + 幂等"，因为幂等的成本远低于维护一套分布式事务。

五、常见追问

• read_committed 会不会读到一半事务的数据？ 不会，它只读到 LSO（Last Stable Offset）之前的数据，未提交事务的消息被跳过。
• 事务超时会怎样？ 超过 transaction.timeout.ms（默认 60 秒）未提交，协调器会中止事务，消息对下游不可见，需要重试整个事务。
• transactional.id 能随便换吗？ 不能。它用于故障恢复时清理悬挂事务，同一业务实例必须稳定；换 ID 会导致旧事务无法被正确中止。
• 幂等生产者就等于只消费一次吗？ 不等于。幂等只保证"同一生产者重试不产生重复"，跨会话、跨分区、消费端重复投递它都管不了。
• Flink 的端到端精确一次是怎么做的？ Checkpoint 对齐 + 两阶段提交 Sink，本质仍是"位点与输出同事务"，只是把事务边界交给了 Checkpoint。

小结

• Kafka 的精确一次只在"读—处理—写—提交位点"全部走 Kafka 事务时成立。
• 位点必须与输出写在同一个事务里，否则回滚后仍会丢。
• 外部副作用无法被 Kafka 事务覆盖，幂等或 Outbox 才是通用解。
• 生产上"至少一次 + 幂等"比"追求精确一次"更常见也更划算。