之前搜索网上kafka事务相关文章，要么不清不楚，要么过于深奥。最近难得一闲，啃了一下kip-98，终于如愿能系统地总结一下kafka事务的原理与实现。

kafka事务

kafka官方设计文档 kip-98。大家可以先阅读我这篇文章，再去啃kip原文。

Kafka 的事务可以看作Kafka 中最难的知识点之一!

一、基础概念

一般而言，消息中间件的消息传输保障有3个层级:

• at most once：至多一次。消息可能会丢失，但绝对不会重复传输
• at least once：最少一次。消息绝不会丢失，但可能会重复传输
• exactly once：恰好一次。每条消息肯定会被传输一次且仅传输一次

Kafka 的消息传输保障机制非常直观。当生产者向 Kafka 发送消息时，一旦消息被成功提交到日志文件，由于多副本机制的存在，这条消息就不会丢失。

Kafka从0.11.0.0版本开始引入了幂等和事务这两个特性，以此来实现EOS（exactly oncesemantics，精确一次处理语义）。

二、幂等

所谓的幂等，简单地说就是对接口的多次调用所产生的结果和调用一次是一致的。生产者在进行重试的时候有可能会重复写入消息，而使用Kafka的幂等性功能之后就可以避免这种情况。

为了实现生产者的幂等性，Kafka为此引入了

• producer id
• 序列号（sequencenumber）这两个概念(在早期的kafka版本中，我们在业务层也实现了类似的header信息来实现业务层幂等。当然这部分kafka原生实现的源码也值得一看，对比看看有何优劣!)

客户端需要开启enable.idempotence为true。

为什么需要幂等？

因为事务开启后必然会有很多的失败重试

三、事务

幂等性并不能跨多个分区运作，而事务可以弥补这个缺陷。事务可以保证对多个分区写入操作的原子性。操作的原子性是指多个操作要么全部成功，要么全部失败，不存在部分成功、部分失败的可能。

1、解决什么问题？

• 多分区原子写入
• consume-transform-produce 原子操作
• 有状态应用多个会话之间的连续性如何理解： 如应用重启后可以接着幂等生产和事务恢复

2、producer和consumer

2.1 producer

• producer和transactionalId建议一一对应，因为一个TransactionalId只能有一个活跃的producer
• 应用实例die的情况下，新启动的实例会接管事务，继续未完成的事务(aborted或committed)

2.2 consumer

consumer 事务语义相对弱一些

• topic是compacted，老数据可能被新版本的数据覆盖
• 事务消息跨多个日志segment，可能其中部分segment过期被删除
• consumer消费可以指定offset消费。10条数据原子一个事务原子写入，但你指定从第9条数据开始消费，则1~8条数据不会被消费到。
• consumer不一定会同时消费所有一个事务生产的topic和分区

3、具体实现

3.1 关键概念

为了支持事务，kafka新增加了如下模块：

• 新协调器：Transaction Coordinator，分配PID、管理事务
• 新内部topic：__transaction_state 持久化存储事务的日志
• 新消息类型： ControlBatch兼容老的ProduceBatch，写到用户的topic，告诉用于之前fetch的数据是否已经commit。详见jira。 之前线上有个案例case，旧版本客户端消费该数据导致故障。
• 新id：transactionalId，唯一识别producer。旧producer挂了的情况下，同id的新producer继续未完成的事务。
• 新的epoch：producer epoch。通过版本号确定合法的事务producer(一个TransactionalId只能有一个活跃的producer)。
• 新的request类型、新的request版本号
• 新的消息格式

3.2 案例与DataFlow

注意4.3和4.4是Consume-Transform-Produce场景独有的步骤，涉及到消费和生产的联动，这里我不多介绍。

步骤:

• 1、查找TransactionCoordinator
• 2、获取并保存PID
• 3、Producer开启事务
• 4、Consume-Transform-Produce(以该场景举例)
  - 4.1 AddPartitionsToTxnRequest：producer告诉事务协调器哪些tp会被写入，事务协调器会将这些信息存储，并启动一个事务的定时器。
  - 4.2 ProduceRequest。 与常规的Produce相比多了PID\\epoch\\sequence number字段
  - 4.3 AddOffsetCommitsToTxnRequest：批量消费和生产时使用到。
  sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
  String consumerGroupId)
  - 4.4 TxnOffsetCommitRequest 与4.3配合
• 5、提交或取消一个事务
  - 5.1 EndTxnRequest：endTransaction \\ abortTransaction
abortTransaction 后 下游消费者读取后会丢弃掉对应的数据
  收到这个请求后，
  1）协调器将往事务日志里面写入PREPARE_COMMIT和PREPARE_ABORT消息。
  2）发送COMMIT或ABORTmarker到用户的数据里面(即5.2)。
  3）事务日志写入COMMITTED和ABORTED消息。
  - 5.2 WriteTxnMarkerRequest事务协调器发给各个tp的leader的broker，broker收到后会往对应的topic里面插入ControlBatch(即marker)。告知该PID之前发送的消息的状态是怎么样的。注意：消费者需要缓存该PID的消息直到收到ControlBatch信息，才直到应该如何处理这些消息(业务处理or丢弃)。
• 5.3 发送最终提交或取消请求ControlBatch(即marker)写到tp里后，事务协调器会往事务日志里面写入COMMITTED和ABORTED消息，事务完成。

四、事务关键参数

1、broker端

• transactional.id.timeout.ms：604800000 (7 days) 事务id在没有更新情况下存活的时间。
• max.transaction.timeout.ms：900000 (15 min) 最大事务的超时时间。

2、producer端

• enable.idempotence： 幂等默认会要求： acks=all retries > 1 max.inflight.requests.per.connection=1.
• transaction.timeout.ms：事务超时时间，60000
• transactional.id：事务id

3、consumer端

• isolation.level：默认read_uncommitted
  • read_uncommitted: 同时消费到committed和uncommitted的消息
  • read_committed：只能消费到非事务消息以及

五、性能测试

不难分析，kafka的事务是相对轻量的，对性能影响相对可控。

kafka-producer-perf-test.sh的 --transactional-id 用于测试并发事务的性能。

潜在性能下降点：单producer写入性能，特别是开启幂等后的性能下降。