当前位置：首页 > news >正文

【kafka系列】Exactly Once语义

news 来源：原创 2025/5/13 14:08:36

1. Exactly-Once语义的定义

2. Kafka实现Exactly-Once的机制

3. 端到端Exactly-Once示例

场景描述

3.1 生产者配置与代码

3.2 消费者配置与代码

4. 异常场景与Exactly-Once保障

场景1：生产者发送消息后宕机

场景2：消费者处理消息后宕机

场景3：Broker宕机

5. 关键实现细节

6. 总结

1. Exactly-Once语义的定义

Exactly-Once（精确一次）语义指：消息从生产到消费的整个生命周期中，每条消息被严格处理且仅处理一次。即使在生产者重试、Broker故障或消费者重启等场景下，也能避免数据重复或丢失。

2. Kafka实现Exactly-Once的机制

Kafka通过以下三部分实现Exactly-Once：

幂等性生产者（Idempotent Producer）

- 确保单分区内消息不重复（通过PID和Sequence Number）。

事务（Transactions）

- 跨分区的原子性写入（通过两阶段提交和事务协调器）。

消费者端去重（Consumer Deduplication）

- 结合事务和外部存储（如数据库）实现端到端精确一次。

3. 端到端Exactly-Once示例

场景描述

一个订单处理系统：

生产者：发送订单支付消息到Topic orders。
消费者：消费消息，扣减用户账户余额，并将结果写入数据库。
要求：订单支付消息必须被精确处理一次（避免重复扣款）。

3.1 生产者配置与代码

// 生产者配置
Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092");
props.put("acks", "all");
props.put("enable.idempotence", "true"); // 开启幂等性
props.put("transactional.id", "order-producer"); // 必须设置事务ID

// 初始化事务
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

try {
    producer.beginTransaction();

    // 发送订单消息到Topic orders
    producer.send(new ProducerRecord<>("orders", "order-1001", "支付100元"));

    // 其他操作（如写入其他Topic）
    producer.send(new ProducerRecord<>("audit_log", "order-1001", "已处理"));

    producer.commitTransaction(); // 提交事务
} catch (Exception e) {
    producer.abortTransaction(); // 中止事务
    throw e;
}

3.2 消费者配置与代码

// 消费者配置
Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092");
props.put("group.id", "order-group");
props.put("isolation.level", "read_committed"); // 仅消费已提交的事务消息

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("orders"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        // 1. 检查订单是否已处理（数据库去重）
        if (!isOrderProcessed(record.key())) {
            // 2. 扣减账户余额
            deductBalance(record.key(), record.value());
            // 3. 记录处理状态到数据库（原子操作）
            markOrderAsProcessed(record.key());
        }
        // 4. 手动提交Offset（事务性提交）
        consumer.commitSync();
    }
}