特性背景

消息事务是指一系列的生产、消费操作可以要么都完成，要么都失败，类似数据库的事务。这个特性在0.10.2的版本是不支持的，从0.11版本开始才支持。华为云DMS率先提供Kafka 1.1.0的专享版服务，支持消息事务特性。

支持事务消息有什么作用？消息事务是实现分布式事务的一种方案，可以确保分布式场景下的数据最终一致性。例如最常用的转账场景，小王 转账到小明，实际操作是小王账户减去相应金额，小明的账户增加相应金额，在分库分表的前提下，2个账户存储在不同的数据库中，这时需要分布式事务才能保证数据库一致性，单个数据库的事务无法保证跨库之间的原子性。如果小王账户先扣钱，再去发送消息到小明所在的数据库去通知增加钱，在没有事务消息的情况下，无论是先扣钱或者先发送通知增加钱，都会有数据不一致的问题，因为无法保证两者的原子性。而有了事务消息，可以保证发送通知与本地事务(扣钱)是一个原子操作，本地事务与发送通知可以同时成功或者同时失败，确保数据一致。

除了数据最终一致性外，还实现了消息Exactly once语义。所谓Exactly once语义是消息传递语义中最难实现的一种，包括At most once：最多一次(不会重复，但是可能丢失数据)； At least once：至少投递一次(不会丢失，但是会导致重复)和Exactly once: 刚好一次(不丢不重)，也即幂等性。Kafka的幂等性可以保证生产只对一个分区实现Exactl once语义，需要多个分区也实现这个语义，还需要引入消息事务确保原子性。

分布式事务介绍

当前系统架构主流是分布式架构与微服务架构，在这种架构下数据源不是单一的数据库，业务逻辑往往需要在多个数据库中实现原子操作，单个数据库中的强大的本地事务无法保证多节点原子操作。 此时需要分布式事务来确保数据的一致性。目前使用较多的分布式事务解决方案有几种：

1、XA事务：两阶段/三阶段提交

XA是由X/Open组织提出的分布式事务的规范。XA规范主要定义了(全局)事务管理器(Transaction Manager)和(局部)资源管理器(Resource Manager)之间的接口。XA接口是双向的系统接口，在事务管理器(Transaction Manager)以及一个或多个资源管理器(Resource Manager)之间形成通信桥梁。实现XA事务的关键是两阶段和三阶段提交协议。

两阶段提交协议(Two-phase Commit，2PC)经常被用来实现分布式事务。一般分为协调器C和若干事务参与者Si两种角色，这里的事务参与者就是具体的数据库，协调器可以和事务参与者在一台机器上，如下图

二阶段提交协议主要包括由2个阶段：第一个阶段为准备阶段(prepare)，第二阶段为提交阶段。准备阶段由事务协调者向事务参与者发送prepare消息，各个参与者处理本地事务但不提交，然后向事务协调者返回事务状态。 提交阶段根据准备阶段各参与者的执行请求，协调者确定事务是提交或者回滚，向各个参与者发送命令。

二阶段提交协议主要的问题是在提交执行过程中，所有的参与者都需要听从协调者的统一调度，期间处于阻塞状态而不能从事其他操作，这样效率及其低下。特别是当协调者发出提交通知到部分参与者后宕机，其他参与者就会阻塞。

针对二阶段提交存在的问题，三阶段提交协议在prepare与commit阶段之间增加一个pre-commit阶段。Prepare阶段只询问参与者而不做事务，而在pre-commit阶段各个参与者才会执行本地事务但不提交。Commit阶段就是直接提交。这样做可以避免二阶段当协调者迟迟没有发出commit或者rollback通知，参与者在超时后可以自行提交或者回滚，避免阻塞事务(这是因为经过了prepare阶段已经确认了各个参与者是可以执行的，最后第三阶段直接执行即可)。 三阶段提交也存在很多问题，也不能完全保证数据一致，完全一致需要用到Paxos算法。

2、TCC补偿性事务解决方案

TCC分别对应Try、Confirm和Cancel三种操作，含义如下：

- Try：预留业务资源

- Confirm：确认执行业务操作，执行事务

- Cancel：取消执行业务操作

TCC解决了跨应用业务操作的原子性问题，在诸如组合支付、账务拆分场景非常实用。TCC实际上把数据库层的二阶段提交上提到了应用层来实现，对于数据库来说是一阶段提交，规避了数据库层的2PC性能低下问题。TCC需要业务提供使用，开发复杂和成本高。

3、事务消息

基于消息中间件的事务消息来完成分布式事务。事务消息可以确保本地执行事务与消息发送是原子的：先发送一条消息到消息中间件，然后执行本地事务，当本地事务成功后再发送提交确认到消息中间件，然后这条消息才能被其他业务消费者所能感知，从而确保原子性。

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发送数据的每个都对应一个从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 左边)。

●写入最终提交或回滚信息

当提交和回滚信息写入数据日子后，事务协调者会往事务日志中写入最终的提交或者终止信息以表示事务已经完成(图5.3)，此时大部分于事务有关系的消息都可以被删除(通过标记后面在日志压缩时会被移除)，我们只需要保留事务ID以及其时间戳即可。

接口

示例