Kafka Exactly-Once语义

约 1819 字大约 6 分钟

kafkaexactly-once

2025-05-31

概述

消息传递语义是消息队列系统中最核心的问题之一。Kafka从0.11版本开始支持Exactly-Once Semantics（EOS），通过幂等生产者（Idempotent Producer）和事务（Transactions）两个机制，实现了端到端的精确一次处理。本文详细解析这些机制的原理与实现。

三种消息语义

At-Most-Once

// At-Most-Once配置
Properties props = new Properties();
props.put("acks", "0");      // 不等待确认
props.put("retries", 0);      // 不重试

// 问题：消息可能丢失
// 场景：Producer发送后网络故障，Broker未收到消息

At-Least-Once

// At-Least-Once配置
Properties props = new Properties();
props.put("acks", "all");     // 等待所有ISR确认
props.put("retries", 3);      // 失败重试

// 问题：消息可能重复
// 场景：Broker已写入消息并同步到ISR，但ACK返回前网络断开
//       Producer超时重试，导致同一消息被写入两次

幂等生产者（Idempotent Producer）

原理

Kafka通过Producer ID（PID）和Sequence Number实现幂等性：

PID：每个Producer实例启动时，Broker分配一个唯一的Producer ID
Sequence Number：Producer为每个 <Topic, Partition> 维护一个单调递增的序列号

配置

Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092");
props.put("enable.idempotence", "true"); // 启用幂等性

// 幂等性隐含的配置要求：
// acks = "all"（自动设置）
// retries > 0（自动设置为Integer.MAX_VALUE）
// max.in.flight.requests.per.connection <= 5（必须）

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

// 使用方式不变，幂等性对应用透明
producer.send(new ProducerRecord<>("topic", "key", "value"));

局限性

幂等生产者只能保证单分区、单会话的幂等性：

单分区：PID+Seq是按Partition维护的，跨分区不保证
单会话：Producer重启后PID会改变，新旧PID之间无法去重

事务（Transactions）

Kafka事务解决了幂等生产者的局限性，支持跨分区、跨会话的原子性写入。

代码示例

Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092");
props.put("enable.idempotence", "true");
props.put("transactional.id", "order-producer-1"); // 事务ID（跨会话唯一）

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

// 初始化事务（每个Producer实例只调用一次）
producer.initTransactions();

try {
    // 开启事务
    producer.beginTransaction();

    // 发送多条消息到不同分区/Topic（原子性）
    producer.send(new ProducerRecord<>("orders", "order-1", orderJson));
    producer.send(new ProducerRecord<>("payments", "pay-1", paymentJson));
    producer.send(new ProducerRecord<>("inventory", "item-1", inventoryJson));

    // 提交Consumer offset到事务中（Consume-Transform-Produce模式）
    Map<TopicPartition, OffsetAndMetadata> offsets = Map.of(
        new TopicPartition("input-topic", 0),
        new OffsetAndMetadata(lastConsumedOffset + 1)
    );
    producer.sendOffsetsToTransaction(offsets, new ConsumerGroupMetadata("my-group"));

    // 提交事务
    producer.commitTransaction();
} catch (ProducerFencedException e) {
    // 另一个相同transactional.id的Producer已启动
    producer.close();
} catch (KafkaException e) {
    // 中止事务
    producer.abortTransaction();
}

Consumer端 read_committed

Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092");
props.put("group.id", "order-consumer-group");

// 关键配置：只读取已提交的事务消息
props.put("isolation.level", "read_committed");
// read_uncommitted: 读取所有消息（包括未提交事务的）
// read_committed: 只读取已提交事务的消息和非事务消息

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

read_committed的工作原理

LSO（Last Stable Offset）：Consumer的read_committed模式下，只能读取到LSO之前的消息。LSO是最早还未完成的事务的第一条消息的offset。这意味着一个长时间未提交的事务会阻塞后续所有消息的消费。

端到端Exactly-Once

Consume-Transform-Produce模式

// 完整的Consume-Transform-Produce示例
public class ExactlyOnceProcessor {

    public void process() {
        KafkaConsumer<String, String> consumer = createConsumer();
        KafkaProducer<String, String> producer = createTransactionalProducer();

        producer.initTransactions();
        consumer.subscribe(List.of("input-topic"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            if (records.isEmpty()) continue;

            producer.beginTransaction();
            try {
                Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();

                for (ConsumerRecord<String, String> record : records) {
                    // Transform
                    String result = transform(record.value());

                    // Produce
                    producer.send(new ProducerRecord<>("output-topic",
                        record.key(), result));

                    // 记录offset
                    offsets.put(
                        new TopicPartition(record.topic(), record.partition()),
                        new OffsetAndMetadata(record.offset() + 1));
                }

                // 将offset提交与消息发送放在同一事务中
                producer.sendOffsetsToTransaction(offsets,
                    consumer.groupMetadata());
                producer.commitTransaction();

            } catch (Exception e) {
                producer.abortTransaction();
                // Consumer会从上次提交的offset重新消费
            }
        }
    }
}

性能影响

模式	吞吐量影响	延迟影响	说明
幂等Producer	~3-5%下降	几乎无	额外的Seq校验开销
事务Producer	~20%下降	增加10-50ms	事务日志写入 + 标记传播
read_committed	可能增加延迟	受LSO限制	长事务会阻塞消费

Kafka的Exactly-Once语义通过幂等生产者和事务机制分层实现。幂等生产者使用PID+Sequence Number解决了单分区内的消息去重问题，成本低且对应用透明。事务机制在此基础上支持跨分区的原子性写入，配合Consumer端的read_committed隔离级别，实现了端到端的Exactly-Once保证。在Consume-Transform-Produce场景中，将offset提交纳入事务是实现端到端EOS的关键。实际使用中需要权衡EOS带来的性能开销，并注意避免长事务阻塞LSO。

贡献者

withesse

更新日志

2026/3/14 13:09

查看所有更新日志

9f6c2-feat: organize wiki content and refresh site setup于 2026/3/14