【干货】Kafka 事务特性分析

特性背景

消息事务是指一系列的生产、消费操做能够要么都完成，要么都失败，相似数据库的事务。这个特性在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语义，须要多个分区也实现这个语义，还须要引入消息事务确保原子性。

分布式事务介绍

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

一、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算法。

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

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

Try：预留业务资源

Confirm：确认执行业务操做，执行事务

Cancel：取消执行业务操做

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

三、事务消息

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

Kafka消息事务

1、基本概念

为了支持事务，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，能够独立与事务使用，便可以只开启幂等但不开启事务。

2、事务流程

以下图所示：

一、查找事务协调者

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

二、获取produce ID

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

●不带transactionID

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

●带transactionID

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

三、启动事务

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

四、消费和生产配合过程

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

●增长分区到事务请求

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

●生产请求

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

●增长消费offset到事务

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

●事务提交offset请求

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

五、提交或回滚事务

用户经过调用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以及其时间戳便可。

接口

示例

欢迎工做一到五年的Java工程师朋友们加入Java架构开发：744677563

本群提供免费的学习指导 架构资料 以及免费的解答

不懂得问题均可以在本群提出来 以后还会有职业生涯规划以及面试指导