【干货】Kafka 事务特性分析（下）_文化 & 方法_李浩

QCon北京「鸿蒙专场」火热来袭！即刻报名，与创新同行~ 了解详情 



 写点什么

Kafka 消息事务

01 基本概念

为了支持事务，Kafka 0.11.0 版本引入以下概念：

1、事务协调者：类似于消费组负载均衡的协调者，每一个实现事务的生产端都被分配到一个事务协调者(Transaction Coordinator)。

2、引入一个内部 Kafka Topic 作为事务 Log：类似于消费管理 Offset 的 Topic，事务 Topic 本身也是持久化的，日志信息记录事务状态信息，由事务协调者写入。

3、引入控制消息(Control Messages)：这些消息是客户端产生的并写入到主题的特殊消息，但对于使用者来说不可见。它们是用来让 broker 告知消费者之前拉取的消息是否被原子性提交。

4、引入 TransactionId：不同生产实例使用同一个 TransactionId 表示是同一个事务，可以跨 Session 的数据幂等发送。当具有相同 Transaction ID 的新的 Producer 实例被创建且工作时，旧的且拥有相同 Transaction ID 的 Producer 将不再工作，避免事务僵死。

5、Producer ID：每个新的 Producer 在初始化的时候会被分配一个唯一的 PID，这个 PID 对用户是不可见的。主要是为提供幂等性时引入的。

6、Sequence Numbler。（对于每个 PID，该 Producer 发送数据的每个< Topic, Partition>都对应一个从 0 开始单调递增的 Sequence Number。

7、每个生产者增加一个 epoch：用于标识同一个事务 Id 在一次事务中的 epoch，每次初始化事务时会递增，从而让服务端可以知道生产者请求是否旧的请求。

8、幂等性：保证发送单个分区的消息只会发送一次，不会出现重复消息。增加一个幂等性的开关 enable.idempotence，可以独立与事务使用，即可以只开启幂等但不开启事务。

02 事务流程

如下图所示：

1、查找事务协调者

生产者会首先发起一个查找事务协调者的请求(FindCoordinatorRequest)。协调者会负责分配一个 PID 给生产者。类似于消费组的协调者。

2、获取 produce ID

在知道事务协调者后，生产者需要往协调者发送初始化 pid 请求(initPidRequest)。这个请求分两种情况：

不带 transactionID

这种情况下直接生成一个新的 produce ID 即可，返回给客户端

带 transactionID

这种情况下，kafka 根据 transactionalId 获取对应的 PID，这个对应关系是保存在事务日志中（上图 2a）。这样可以确保相同的 TransactionId 返回相同的 PID，用于恢复或者终止之前未完成的事务。

3、启动事务

生产者通过调用 beginTransaction 接口启动事务，此时只是内部的状态记录为事务开始，但是事务协调者认为事务开始只有当生产者开始发送第一条消息才开始。

4、消费和生产配合过程

这一步是消费和生成互相配合完成事务的过程，其中涉及多个请求：

增加分区到事务请求

当生产者有新分区要写入数据，则会发送 AddPartitionToTxnRequest 到事务协调者。协调者会处理请求，主要做的事情是更新事务元数据信息，并把信息写入到事务日志中（事务 Topic）。

生产请求

生产者通过调用 send 接口发送数据到分区，这些请求新增 pid，epoch 和 sequence number 字段。

增加消费 offset 到事务

生产者通过新增的 snedOffsets ToTransaction 接口，会发送某个分区的 Offset 信息到事务协调者。协调者会把分区信息增加到事务中。

事务提交 offset 请求

当生产者调用事务提交 offset 接口后，会发送一个 TxnOffsetCommitRequest 请求到消费组协调者，消费组协调者会把 offset 存储在__consumer-offsets Topic 中。协调者会根据请求的 PID 和 epoch 验证生产者是否允许发起这个请求。消费 offset 只有当事务提交后才对外可见。

5、提交或回滚事务

用户通过调用 commitTransaction 或 abortTranssaction 方法提交或回滚事务。

EndTxnRequest

当生产者完成事务后，客户端需要显式调用结束事务或者回滚事务。前者会使得消息对消费者可见，后者会对生产数据标记为 Abort 状态，使得消息对消费者不可见。无论是提交或者回滚，都是发送一个 EndTnxRequest 请求到事务协调者，写入 PREPARE_COMMIT 或者 PREPARE_ABORT 信息到事务记录日志中(5.1a)。

WriteTxnMarkerRequest

这个请求是事务协调者向事务中每个 TopicPartition 的 Leader 发送的。每个 Broker 收到请求后会写入 COMMIT(PID)或者 ABORT(PID)控制信息到数据日志中(5.2a)。

这个信息用于告知消费者当前消息是哪个事务，消息是否应该接受或者丢弃。而对于未提交消息，消费者会缓存该事务的消息直到提交或者回滚。

这里要注意，如果事务也涉及到__consumer_offsets，即该事务中有消费数据的操作且将该消费的 Offset 存于__consumer_offsets 中，Transaction Coordinator 也需要向该内部 Topic 的各 Partition 的 Leader 发送 WriteTxnMarkerRequest 从而写入 COMMIT(PID)或 COMMIT(PID)控制信息(5.2a 左边)。