流处理系统中的 Exactly-Once 语义：基于两阶段提交与幂等写入的工程实现

流处理系统中的 Exactly-Once 语义：基于两阶段提交与幂等写入的工程实现

一、"至少一次"到"精确一次"的质变

流处理中，"至少一次"（At-Least-Once）语义意味着同一事件可能被处理多次——下游需要有幂等性兜底。"精确一次"（Exactly-Once）语义保证每个事件在处理结果中出现且仅出现一次。从外部观察者的视角，就像事件恰好被处理了一次。

这个保证的实现远比字面描述复杂。一个典型场景：Kafka Consumer 消费消息 → 流处理算子转换 → 写入下游数据库。以下三种故障模式都会导致语义违背：

1. 处理成功但提交偏移量失败：消息被处理后写入了数据库，但 Consumer 在提交 Kafka Offset 前崩溃。重启后由于 Offset 未更新，消息被重新消费——导致数据库中有两条相同记录。
2. 偏移量提交成功但处理失败：Consumer 提交了 Offset 后崩溃，消息已被标记为消费但处理结果未写入数据库——消息丢失。
3. 下游写入失败后的重试：消息处理后写入数据库超时，重试时数据库第一次写入可能实际已成功——导致数据重复。

两阶段提交（2PC）是解决这个问题的经典方案：将"消息消费偏移量提交"和"下游写入"作为一个原子事务，要么都成功，要么都失败。不是数据库的 2PC，而是将 Kafka Offset 和下游写入放在同一个逻辑事务中。

二、Exactly-Once 的两种实现路径

方案 A——两阶段提交：

Pre-Commit 阶段：将数据写入下游数据库，但此时标记为"未提交"状态（或写入临时表）。同时将当前 Kafka Offset 暂存到外部存储（如状态后端）。

Commit 阶段：提交下游数据库的事务（将临时数据标记为有效），然后提交 Kafka Offset。如果 Commit 阶段失败，需要根据暂存的 Offset 回滚——删除临时数据，从暂存 Offset 重新消费。

优点：不强依赖下游的幂等性——即使下游不支持幂等写入（如发送邮件、推送通知），也能保证 Exactly-Once。
缺点：引入了外部状态存储（Offset 暂存），且 Commit 阶段的延迟增加了端到端延迟。

方案 B——幂等写入：

核心思路：如果下游操作是幂等的，那么即使重复执行也不产生副作用。对于数据库写入，使用INSERT ... ON CONFLICT (id) DO NOTHING或UPSERT语义。对于 Kafka 写入，使用事务性 Producer（transactional.id+sendOffsetsToTransaction）。

优点：实现简单，无需外部状态存储。与 At-Least-Once 架构兼容——只需要增强下游的幂等性。
缺点：不是所有下游都支持幂等写入。像"发送短信""调用支付接口"这类操作天然非幂等，虽然可以通过唯一请求 ID 实现业务幂等，但复杂度更高。

三、基于 Kafka 事务的 Exactly-Once 实现

use rdkafka::{ consumer::{Consumer, StreamConsumer, CommitMode}, producer::{FutureProducer, FutureRecord}, message::{BorrowedMessage, OwnedHeaders}, ClientConfig, TopicPartitionList, Offset, }; use rdkafka::types::RDKafkaErrorCode; use std::collections::HashMap; use std::time::Duration; /// Exactly-Once 处理上下文 pub struct ExactlyOnceContext { /// 事务 ID 前缀 —— 相同前缀的 Producer 共享事务状态 transactional_id: String, /// Kafka 事务性 Producer producer: FutureProducer, /// Kafka Consumer consumer: StreamConsumer, /// 状态后端 —— 用于暂存 Offset（方案A: 2PC） /// 生产环境应替换为 RocksDB 或远程 KV 存储 offset_store: HashMap<i32, i64>, } impl ExactlyOnceContext { /// 初始化 —— 创建事务性 Producer 和 Consumer pub fn new( transactional_id: &str, brokers: &str, group_id: &str, input_topic: &str, ) -> Result<Self, KafkaError> { // 事务性 Producer 配置 let producer: FutureProducer = ClientConfig::new() .set("bootstrap.servers", brokers) .set("transactional.id", transactional_id) // 事务超时时间（最大允许的事务持续时间） .set("transaction.timeout.ms", "60000") // 60s，超时后 Kafka 自动中止事务 // 启用幂等性（事务性 Producer 自动启用幂等） .set("enable.idempotence", "true") .create()?; // 初始化事务 —— 必须在使用前调用 // init_transactions 向 Kafka 事务协调器注册 producer.init_transactions(Duration::from_secs(30))?; // Consumer 配置 let consumer: StreamConsumer = ClientConfig::new() .set("bootstrap.servers", brokers) .set("group.id", group_id) // 关闭自动偏移量提交 —— 由事务控制提交时机 .set("enable.auto.commit", "false") // 隔离级别：只读取已提交的消息 .set("isolation.level", "read_committed") .create()?; // 订阅 Topic consumer.subscribe(&[input_topic])?; Ok(Self { transactional_id: transactional_id.to_string(), producer, consumer, offset_store: HashMap::new(), }) } /// 事务性处理单条消息 /// /// 保证：消息处理 + 结果写入 + Offset 提交 = 原子操作 pub async fn process_with_transaction<F, Fut>( &mut self, msg: &BorrowedMessage<'_>, process_fn: F, ) -> Result<(), KafkaError> where F: FnOnce(&[u8]) -> Fut, Fut: std::future::Future<Output = Result<Option<Vec<u8>>, String>>, { // ===== 1. 开始事务 ===== self.producer.begin_transaction()?; let payload = msg.payload().unwrap_or(&[]); // ===== 2. 执行业务逻辑 ===== match process_fn(payload).await { Ok(Some(result)) => { // 3a. 写入结果到下游 Kafka Topic let record = FutureRecord::to("output-topic") .payload(&result) .key(msg.key().unwrap_or(&[])); // send 操作在事务上下文中 —— Kafka Broker 暂存但不立即可见 self.producer.send(record, Duration::from_secs(5)) .await .map_err(|(e, _)| KafkaError::Produce(e.to_string()))?; } Ok(None) => { // 过滤消息：不需要产生输出 } Err(e) => { // 业务处理失败 → 中止事务 // 消息不会被标记为已消费，下次重启后重新处理 self.producer.abort_transaction()?; return Err(KafkaError::Processing(e)); } } // ===== 3. 构造 Offset 提交 ===== // 方案 A (2PC): 将当前 Offset 暂存 // 方案 B (幂等): 直接提交 Offset let partition = msg.partition(); let offset = msg.offset(); let mut tpl = TopicPartitionList::new(); tpl.add_partition_offset( msg.topic(), partition, Offset::Offset(offset + 1), // Offset 是消费位置 + 1（下一条消息的位置） )?; // 存储 Offset 用于崩溃恢复 self.offset_store.insert(partition, offset); // ===== 4. 发送 Offset 到事务 ===== // send_offsets_to_transaction 将 Consumer Offset 与当前事务绑定 // 事务提交时，Offset 一同被持久化 self.producer.send_offsets_to_transaction( &tpl, // consumer_group_metadata 需要从 Consumer 获取 // 但 rdkafka 的 Rust 绑定对此支持不完整 // 实际代码需要将 consumer 的 group metadata 传给 producer &rdkafka::consumer::ConsumerGroupMetadata::new("group-id".to_string()), Duration::from_secs(10), )?; // ===== 5. 提交事务 ===== // 原子操作：Offset 提交 + 所有 Producer send 的结果持久化 // 如果此处崩溃，Broker 会在 transaction.timeout.ms 后自动中止 self.producer.commit_transaction(Duration::from_secs(10))?; Ok(()) } /// 崩溃恢复：从暂存的 Offset 消费未确认的消息 pub fn recover(&mut self) -> Result<(), KafkaError> { // 读取最后一次暂存的 Offset if let Some((&partition, &offset)) = self.offset_store.iter().last() { // 回退到该 Offset —— 重新消费未确认的消息 // 由于发送到下游的消息在事务中止时被撤销， // 重新消费不会导致重复 // 注意：实际实现需要将所有 TopicPartition 都回退 println!("Recovering from partition {} offset {}", partition, offset); } Ok(()) } } /// 幂等写入方案：利用数据库的唯一约束 pub struct IdempotentWriter { db_pool: sqlx::PgPool, } impl IdempotentWriter { /// 幂等写入 —— INSERT ... ON CONFLICT DO NOTHING /// /// 每个事件生成唯一 ID（基于 Topic + Partition + Offset） /// 数据库中 id 列有 UNIQUE 约束，重复写入被静默忽略。 /// /// 这样即使消息被重复处理，下游也仅有一条记录。 pub async fn write_event( &self, topic: &str, partition: i32, offset: i64, payload: &[u8], ) -> Result<(), sqlx::Error> { // 生成全局唯一事件 ID let event_id = format!("{}:{}:{}", topic, partition, offset); sqlx::query( "INSERT INTO events (id, topic, partition, offset, payload, created_at) \ VALUES ($1, $2, $3, $4, $5, NOW()) \ ON CONFLICT (id) DO NOTHING" ) .bind(&event_id) .bind(topic) .bind(partition) .bind(offset) .bind(payload) .execute(&self.db_pool) .await?; Ok(()) } /// UPSERT 变体：如果记录存在则更新 pub async fn upsert_event( &self, event_id: &str, status: &str, ) -> Result<(), sqlx::Error> { sqlx::query( "INSERT INTO events (id, status, updated_at) \ VALUES ($1, $2, NOW()) \ ON CONFLICT (id) DO UPDATE SET status = $2, updated_at = NOW()" ) .bind(event_id) .bind(status) .execute(&self.db_pool) .await?; Ok(()) } } #[derive(Debug)] pub enum KafkaError { Client(rdkafka::error::KafkaError), Produce(String), Processing(String), } impl From<rdkafka::error::KafkaError> for KafkaError { fn from(e: rdkafka::error::KafkaError) -> Self { KafkaError::Client(e) } }

关键设计决策：

• transaction.timeout.ms = 60000：事务持续时间超过此值，Kafka Broker 自动中止事务。这为崩溃后的事务清理提供了兜底——即使 Producer 崩溃后没有调用abort_transaction，Broker 也会在超时后自动中止。
• isolation.level = read_committed：Consumer 只读取已提交事务的消息。这保证了 Consumer 不会读到其他 Producer 已发送但事务尚未提交的消息——避免读到后续被回滚的脏数据。
• send_offsets_to_transaction的语义：它将 Consumer Offset 的提交绑定到 Producer 的事务中。事务提交 → Offset 提交成功。事务中止 → Offset 不更新，消息重新消费。
• INSERT ... ON CONFLICT DO NOTHING利用数据库唯一约束实现幂等性：重复写入的资源开销仅为一次索引查找（微秒级），远小于两阶段提交中的事务协调开销。

四、Exactly-Once 的适用边界与权衡

适用场景：

• 支付、计费、库存扣减等对准确性有严格要求的系统。
• Kafka 到数据库的流式 ETL——需要保证每条源消息在目标表中恰好出现一次。
• 跨系统数据同步（如 MySQL Binlog → Elasticsearch），避免数据重复导致的索引膨胀。

不适用场景：

• 允许少量重复的分析类处理（误差异常 < 0.01%）。增加 Exactly-Once 保障的延迟开销对近实时分析场景不划算。
• 下游系统不支持事务或不提供幂等 API。S3 的 PutObject 是幂等的，但 SQS 的 SendMessage 不是（可能产生重复消息）。
• 超低延迟要求的场景（< 10ms）。两阶段提交增加的延迟在 5-20ms（取决于 Broker 的往返时间）。

主要权衡：

1. 2PC vs 幂等写入：2PC 对下游无要求但实现复杂，引入外部状态存储的维护成本。幂等写入简单但依赖下游系统的能力。实际项目中两者的组合最常见——2PC 覆盖不可幂等的操作，幂等写入覆盖可幂等的操作。
2. 事务超时的设置：transaction.timeout.ms越大，事务可处理的业务逻辑越复杂（如需要调用外部 API），但崩溃后未中止事务所占用的 Broker 资源时间越长。60s 是官方推荐的平衡点。
3. 吞吐量影响：事务性 Producer 的吞吐量比普通 Producer 低 20-30%（每条消息需要额外的协调开销）。在高吞吐场景下，通常将多条消息批处理到同一个事务中。

五、总结

1. Exactly-Once 语义的本质是将"消息消费偏移量提交"和"下游写入"作为一个原子操作。
2. 两阶段提交（2PC）方案通过 Pre-Commit + Commit 实现原子性，适用于下游不可幂等的场景。
3. 幂等写入方案通过唯一 ID + 数据库 UPSERT 实现重复执行无副作用，实现更简单。
4. Kafka 事务性 Producer 将 Producer send 和 Consumer Offset 提交绑定为原子操作，是实现 Exactly-Once 的基础设施。
5. isolation.level = read_committed确保 Consumer 不读取未提交事务的脏数据，是端到端 Exactly-Once 的必要配置。