Kafka详解（三）：服务端与客户端详解（时间轮、延时操作、控制器、分区分配策略、再均衡、事务）_kafka延时生产

六、深入服务端

1、协议设计

Kafka自定义了一组基于TCP的二进制协议，只要遵守这组协议的格式，就可以向Kafka发送消息，也可以从中拉取消息。在Kafka2.0.0中，一共包含了43种协议类型，每种协议类型都有对应的请求和响应，它们都遵循特定的协议模式。每种类型的Rquest都包含相同结构的协议请求头和不同结构的协议请求体

协议请求头中包含4个域：api_key、api_version、correlation_id和client_id

每种类型的Response也包含相同结构的协议响应头和不同结构的响应体

协议响应头中只有一个correlation_id

2、时间轮

Kafka的延时操作是基于时间轮的概念自定义实现了一个用于延时功能的定时器。JDK中Timer和DelayQueue的插入和删除操作的平均时间复杂度为$O(nlogn)$，而基于时间轮可以将插入和删除操作的时间复杂度都降为$O(1)$

Kafka中的时间轮是一个存储定时任务的环形队列，底层采用数组实现，数组中的每个元素都可以存放一个定时任务列表（TimerTaskList）。TimerTaskList是一个环形的双向链表，链表中的每一项表示的都是定时任务项，其中封装了真正的定时任务

时间轮由多个时间格组成，每个时间格代表当前时间轮的基本时间跨度（tickMs）。时间轮的时间格个数是固定的，可用wheelSize来表示，那么整个时间轮的总体时间跨度（interval）可以通过公式tickMs*wheelSize计算得出。时间轮还有一个表盘指针（currentTime），用来表示时间轮当前所处的时间，currentTime是tickMs的整数倍。currentTime可以将整个时间轮划分为到期部分和未到期部分，currentTime当前指向的时间格也属于到期部分，表示刚到到期，需要处理此时间格所对应的TimerTaskList中的所有任务

若时间轮的tickMs为1ms且wheelSize等于20，可以计算得出总体时间跨度interval为20ms。初始情况下表盘指针currentTime指向时间格0，此时有一个定时为2ms的任务插进来会存放到时间格为2的TimerTaskList中。随着时间推移，指针currentTime不断向前推进，过了2ms之后，当到达时间格2时，就需要将时间格2对应的TimerTaskList中的任务进行相应的到期操作。此时若有一个定时为8ms的任务插进来，则会存放到时间格10中，currentTime再过8ms后会指向时间格10.如果同时有一个定时为19ms的任务插进来，新来的TimerTaskEntry会复用原来的TimerTaskList，所以它会插入原本已经到期的时间格1。整个时间轮的总体跨度是不变的，随着指针currentTime的不断推进，当前时间轮所能处理的时间段也在不断后移，总体时间范围在currentTime和currentTime+interval之间

Kafka引入了层级时间轮的概念，当任务的到期时间超过了当前时间轮所表示的时间范围时，就会尝试添加到上层时间轮中

上图中，第一层的时间轮tickMs=1ms、wheelSize=20、interval=20ms。第二层的时间轮的tickMs为第一层时间轮的interval，即20ms。每一层时间轮的wheelSize是固定的，都是20，那么第二层的时间轮的总体时间跨度interval为400ms。这个400ms也是第三层的tickMs的大小，第三层的时间轮的总体时间跨度为8000ms

450ms的定时任务第一层时间轮不能满足条件，所以就升级到第二层时间轮中，第二层时间轮也无法满足条件，所以又升级到第三层时间轮中，最终被插入第三层时间轮中时间格1所对应的TimerTaskList。在到期时间为[400ms,800ms)区间内的多个任务都会被放入第三层时间轮的时间格1，时间格1对应的TimerTaskList的超时时间为400ms。随着时间的推移，当此TimerTaskList到期之时，原本定时为450ms的任务还剩下50ms的时间，还不能执行这个任务的到期操作。这时候会做一个时间轮降级的操作，会将整个剩余时间为50ms的定时任务重新提交到层级时间轮中，此时第一层时间轮的总体时间跨度不够，而第二层足够，所以该任务被放到第二层时间轮到期时间为[40ms,60ms)的时间格中。再经历40ms之后，此时这个任务还剩10ms，还是不能立即执行到期操作。所以还要再有一次时间轮的降级，此任务被添加到第一层时间轮到期时间为[10ms,11ms)的时间格中，之后再经历10ms后，此任务真正到期，最终执行相应的到期操作

• 时间轮在创建的时候以当前系统时间为第一层时间轮的起始时间
• 时间轮中的每个双向环形链表TimerTaskList都会有一个哨兵节点，作为第一个节点，它的值域不存储任何东西，只是为了操作的方便而引入的
• 除了第一层时间轮，其余高层时间轮的起始时间都设置为创建此层时间轮前面第一轮的currentTime。第一层的currentTime都必须是tickMs的整数倍
• Kafka中的定时器只需持有时间轮的第一层时间轮的引用，并不会直接持有其他高层的时间轮，但每一层时间轮都会有一个引用指向更高一层的应用，以此层级调用可以实现定时器间接持有各个层级时间轮的引用

Kafka中的定时器借用了JDK中的DelayQueue来协助推进时间轮。对于每个使用到的TimerTaskList（费哨兵节点的定时任务项TimerTaskEntry对应的TimerTaskList）都加入DelayQueue。DelayQueue会根据TimerTaskList对应的超时时间expiration来排序，最短expiration的TimerTaskList会被排在DelayQueue的队头。Kafka中会有一个线程（ExpiredOperationReaper）来获取DelayQueue中到期的任务列表。当线程获取DelayQueue中超时的任务列表TimerTaskList之后，即可以根据TimerTaskList的expiration来推进时间轮的时间，也可以就获取的TimerTaskList执行相应的操作，对里面的TimerTaskEntry该执行过期操作的就执行过期操作，该降级时间轮的就降级时间轮

Kafka中的时间轮专门用来执行插入和删除TimerTaskEntry的操作，而DelayQueue专门负责时间推进的任务

DelayQueue中第一个超时任务列表的expiration为200ms，第二个超时任务为840ms，这里获取DelayQueue的队头只需要$O(1)$的时间复杂度（获取之后DelayQueue内部才会再次切换出新的队头）。如果采用每秒定时推进，那么获取第一个超时的任务列表时执行的200次推进中有199次属于空推进，而获取第二个超时任务时有需要执行639次空推进，这样会无故空耗机器的性能资源

3、延时操作

1）、延时生产

如果在使用生产者客户单发送消息的时候将acks参数设置为-1，那么就意味着需要等待ISR集合中的所有副本都确认收到消息之后才能正确地收到响应的结果，或者捕获超时异常。在将消息写入leader副本的本地日志文件之后，Kafka会创建一个延时的生产操作，用来处理消息正常写入所有副本或超时的情况，以返回相应的响应结果给客户端

随着follower副本不断地与leader副本进行消息同步，进而促使HW进一步增长，HW每增长一次都会检测是否能够完成此次延时生产操作，如果可以就执行以此返回响应结果给客户端；如果在超时时间内始终无法完成，则强制执行

延时操作创建之后会被加入延时操作管理器来做专门的处理。延时操作有可能会超时，每个延时操作管理器都会配备一个定时器来做超时管理，定时器的底层就是采用时间轮实现的。时间轮的轮转是靠线程ExpiredOperationReaper来驱动的，这里ExpiredOperationReaper由延时操作管理器启动的。定时器、ExpiredOperationReaper和延时操作管理器都是一一对应的。延时操作需要支持外部事件的触发，所以还要配备一个监听池来负责监听每个分区的外部事件——查看是否有分区的HW发生了增长。ExpiredOperationReaper不仅可以推进时间轮，还会定期清理监听池中已完成的延时操作

如果客户端设置的acks参数不为-1，或者没有成功的消息写入，那么就直接返回结果给客户端，否则就需要创建延时生产操作并存入延时操作管理器，最终要么由外部事件触发，要么超时触发而执行

2）、延时拉取

两个follower副本都已经拉取到了leader副本的最新位置，此时又向leader副本发送拉取请求，而leader副本没有新的消息写入，这时Kafka选择了延时操作来处理这种情况。Kafka在处理拉取请求时，会先去读一次日志文件，如果收集不到足够多（fetchMinBytes，由参数fetch.min.bytes配置，默认值为1）的的消息，那么就会创建一个延时拉取操作以等待拉取到足够数量的消息。当延时拉取操作执行时，会再读取一次日志文件，然后将拉取结果返回给follower副本。延时拉取操作也会有一个专门的延时操作管理器负责管理，大体和延时生产相同。如果拉取进度一直没有追赶上leader副本，那么在拉取leader副本的消息时一般拉取的消息大小都会不小于fetchMinBytes，这样Kafka也就不会创建相应的延时拉取操作，而是立即返回拉取结果

4、控制器

1）、控制器的选举及异常恢复

Kafka中的控制器选举工作依赖于ZooKeeper，成功竞选为控制器的broker会在ZooKeeper中创建/controller这个临时节点，此临时节点的内容参考如下：

[zk: localhost:2181(CONNECTED) 0] get /controller
{"version":1,"brokerid":1,"timestamp":"1558621064185"}
1
2

其中version在目前版本中固定为1，brokerid表示成为控制器的broker的id编号，timestamp表示竞选成为控制器时的时间戳

在任意时刻，集群中有且仅有一个控制器。每个broker启动的时候会去尝试读取/controller节点的brokerid的值，如果读取到brokerid的值不为-1，则表示已经有其他broker节点成功竞选为控制器，所以当前broker就会放弃竞选；如果ZooKeeper中不存在/controller节点，或者这个节点中的数据异常，那么就会尝试去创建/controller节点。当前broker去创建节点的时候，也有可能其他broker同时去尝试创建这个节点，只有创建成功的那个broker才会成为控制器，而创建失败的broker竞选失败。每个broker都会在内存中保存当前控制器的brokerid值，这个值可以标记为activeControllerId

ZooKeeper中还有一个与控制器有关的/controller_epoch节点，这个节点是持久节点，节点中存放的是一个整型的controller_epoch值。controller_epoch用于记录控制器发生变更的次数，即记录当前的控制器是第几代控制器，可以称为控制器的纪元

controller_epoch的初始值为1，当控制器发生变更时，每选出一个新的控制器就将该字段值加1。每个和控制器交互的请求都会携带controller_epoch这个字段，如果请求的controller_epoch值小于内存中的controller_epoch值，则认为这个请求是向已经过期的控制器所发送的请求，被认定为无效请求。如果请求的controller_epoch值大于内存的controller_epoch值，那么说明已经有新的控制器当选了。Kafka通过controller_epoch来保证控制器的唯一性

控制器在选举成功之后会读取ZooKeeper中各个节点的数据来初始化上下文信息（ControllerContext），并且需要管理这些上下文信息。不管是监听器触发的事件，还是定时任务触发的事件，或者是其他事件都会读取或更新控制器中的上下文信息，那么就会涉及多线程间的同步。针对这一现象，Kafka的控制器使用单线程基于事件队列的模型，将每个事件都做一层封装，然后按照事件发生的先后顺序暂存到LinkedBlockingQueue中，最后使用一个专用的线程（ControllerEventThread）按照FIFO的原则顺序处理各个事件，这样不需要锁机制就可以在多线程间维护线程安全

在目前的新版本的设计中，只有Kafka Controller在ZooKeeper上注册相应的监听器，其他的broker极少需要再监听ZooKeeper中的数据变化。不过每个broker还是会对/controller节点添加监听器，以此来监听此节点的数据变化

当/controller节点的数据发生变化时，每个broker都会更新自身内存中保存的activeControllerId。如果broker在数据变更前是控制器，在数据变更后自身的brokerid值与新的activeControllerId值不一致，那么就需要退位，关闭相应的资源，比如关闭状态机、注销相应的监听器等。有可能控制器由于异常而下线，造成/controller这个临时节点被自动删除；也有可能是其他原因将此节点删除了

当/controller节点被删除时，每个broker都会进行选举，如果broker在节点被删除前是控制器，那么在选举前还需要有一个退位的动作。如果有特殊需要，则可以手动删除/controller节点来触发新一轮的选举。当然关闭控制器所对应的broker，以及手动向/controller节点写入新的brokerid的所对应的数据，同样可以触发新一轮选举

2）、优雅关闭

先获取Kafka的服务进程号PIDS，使用kill -s TERM $PIDS或kill -15 $PIDS的方式来关闭进程

Kafka服务入口程序中有一个名为kafka-shutdown-hock的关闭钩子，待Kafka进程捕获终止信号的时候会执行这个关闭钩子中的内容，其中除了正常关闭一些必要的资源，还会执行一个控制关闭（ControlledShutdown）的动作。使用ControlledShutdown的方式关闭Kafka有两个优点：一是可以让消息完全同步到磁盘上，在服务下次重新上线时不需要进行日志的恢复操作；二是ControlledShutdown在关闭服务之前，会对其上的leader副本进行迁移，这样就可以减少分区的不可用时间

需要成功执行ControlledShutdown动作需要将参数controlled.shutdown.enable（默认值为true）的值设置为true。ControlledShutdown动作如果执行不成功还会重试执行，这个重试的动作由参数controlled.shutdown.max.retries配置，默认为3次，每次重试的间隔由参数controlled.shutdown.retry.backoff.ms设置，默认为5000ms

上图中，有两个broker，其中待关闭的broker的id为x，Kafka控制器所对应的broker的id为y。待关闭的broker在执行ControlledShutdown动作时首先与Kafka控制器建立专门连接，然后发送ControlledShutdownRequest请求，ControlledShutdownRequest请求只有一个brokerId字段，这个brokerId字段的值设置为自身的brokerId的值，即x

Kafka控制器在收到ControlledShutdownRequest请求之后会将与待关闭broker有关联的所有分区进行专门的处理，有关联是指分区中有副本位于这个待关闭的broker之上（这里会有若干次交互）

如果这些分区的副本数大于1且leader副本位于待关闭broker之上，那么需要实施leader副本的迁移及新的ISR的变更。具体的选举分配的方案由专用的选举器ControlledShutdownLeaderSelector提供

如果这些分区的副本数只是大于1，leader副本并不位于待关闭broker之上，那么就由Kafka控制器来指导这些副本的关闭。如果这些分区的副本数只是为1，那么这个副本的关闭动作会在整个ControlledShutdown动作执行之后由副本管开启来具体实施

对于分区的副本数大于1且leader副本位于待关闭broker上的这种情况，如果在Kafka控制器处理之后leader副本还没有成功迁移，那么会将这些没有成功迁移leader副本的分区记录下来，并写入ControlledShutdownResponse的响应

待关闭的broker在收到ControlledShutdownResponse响应之后，需要判断整个ControlledShutdown动作是否执行成功，以此来进行可能的重试或继续执行接下来的关闭资源的动作。执行成功的标准时ControlledShutdownResponse中error_code字段值为0，并且partitions_remaining数组字段为空

3）、分区leader的选举

分区leader副本的选举由控制器负责具体实施。当创建分区或分区上线的时候都需要执行leader的选举动作，对应的选举策略为OfflinePartitionLeaderElectionStrategy。这种策略的基本思路是按照AR集合中副本的顺序查找第一个存货的副本，并且这个副本在ISR集合中。一个分区的AR集合在分配的时候就被指定，并且只要不发生重分配的情况，集合内部副本的顺序是保持不变的，而分区的ISR集合中副本的顺序可能会改变

如果ISR集合中没有可用的副本，那么此时还要再检查一下所配置的unclean.leader.election.enable参数（默认值为flase）。如果这个参数配置为true，那么表示允许从非ISR列表中选举leader，从AR列表中找到第一个存活的副本即为leader

当分区进行重分配的时候也需要执行leader的选举动作，对应的选举策略为ReassignPartitionLeaderElectionStrategy。从重分配的AR列表中找到第一个存活的副本，且这个副本在目前的ISR列表中

当发生优先副本的选举时，直接将优先副本设置为leader即可，AR集合中的第一个副本即为优先副本

当某节点被优雅地关闭，位于这个节点的leader副本都会下线，需要执行leader的选举。选举策略：从AR列表中找到第一个存活的副本，且这个副本在目前的ISR列表中，与此同时还要确保这个副本部委于正在被关闭的节点上

5、参数解密

1）、broker.id

在Kafka集群中，每个broker都有唯一的id值来区分彼此。broker在启动时会在ZooKeeper中/brokers/ids路径下创建一个以当前brokerId为名称的虚节点，broker的健康状态检查就依赖于此虚节点。当broker下线时，该虚节点会自动删除，其他broker节点或客户端通过判断/brokers/ids路径下是否有此broker的brokerId节点来确定该broker的健康状态

可以通过broker端的配置文件config/server.properties里的broker.id参数来配置brokerId，默认情况下broker.id值为-1。在Kafka中，brokerId值必须大于等于0才有可能正常启动，还可以通过meta.properties文件或自动生成功能来实现

#Tue May 21 21:36:16 CST 2019
version=0
broker.id=0
1
2
3

meta.properties文件中记录了与当前Kafka版本对应的一个version字段，不过目前只有一个为0的固定值。还有一个broker.id，即brokerId值。broker在成功启动之后在每个日志根目录下都会有一个meta.properties文件

meta.properties文件与broker.id的关联如下：

1）如果log.dir或log.dirs中配置了多个日志根目录，这些日志根目录中的meta.properties文件所配置的broker.id不一致则会抛出InconsistentBrokerIdException的异常

2）如果config/server.properties配置文件里配置的broker.id的值和meta.properties文件里的broker.id值不一致，那么同样会抛出InconsistentBrokerIdException的异常

3）如果config/server.properties配置文件中未配置broker.id的值，那么就以meta.properties文件中的broker.id值为准

4）如果没有meta.properties文件，那么在获取合适的broker.id值之后会创建一个新的meta.properties文件并将broker.id值存入其中

Kafka提供了另外两个broker端参数：broker.id.generation.enable和reserved.broker.max.id来配合生成新的brokerId。broker.id.generation.enable参数用来配置是否开启自动生成brokerId的功能，默认情况下为true，即开启此功能。自动生成的brokerId有一个基准值，即自动生成的brokerId必须超过这个基准值，这个基准值通过reserved.broker.max.id参数配置，默认值为1000。也就是说，默认情况下自动生成的brokerId从1001开始

自动生成的brokerId的原理是先往ZooKeeper中的/brokers/seqid节点中写入一个空字符串，然后获取返回的Stat信息中的version值，进而将dataVersion的值和reserved.broker.max.id参数配置的值相加。先往节点中写入数据再获取Stat信息，这样可以确保返回的dataVersion值大于0，进而就可以确保生成的brokerId值大于reserved.broker.max.id参数配置的值，符合非自动生成的broker.id的值在[0,reserved.broker.max.id]区间设定

2）、bootstrap.servers

一般可以简单地认为bootstrap.servers这个参数所要指定的就是将要连接的Kafka集群的broker地址列表。不过从深层次的意义上来讲，这个参数配置的是用来发现Kafka集群元数据信息的服务地址

客户端KafkaProducer1与Kafka Cluster直连是客户端给我们的既定印象，而事实上客户端连接Kafka集群要经历以下3个过程：

1）客户端KafkaProducer2与bootstrap.servers参数所指定的Server连接，并发送MetadataRequest请求来获取集群的元数据信息

2）Server在收到MetadataRequest请求之后，返回MetadataResponse给KafkaProducer2，在MetadataResponse中包含了集群的元数据信息

3）客户端KafkaProducer2收到MetadataResponse之后解析出其中包含的集群元数据信息，然后与集群中的各个节点建立连接，之后就可以发送消息了

七、深入客户端

1、分区分配策略

Kafka提供了消费者客户端参数partition.assignment.strategy来设置消费者与订阅主题之间的分区分配策略。默认情况下，此参数的值为org.apache.kafka.clients.consumer.RangeAssignor，即采用RangeAssignor分配策略

1）、RangeAssignor分配策略

RangeAssignor分配策略的原理是按照消费者总数和分区总数进行整除运算来获得一个跨度，然后将分区按照跨度进行平均分配，以保证分区尽可能均匀地分配给所有的消费者。对于每一个主题，RangeAssignor策略会将消费组内所有订阅这个主题的消费者按照名称的字典序排序，然后为每个消费者划分固定的分区范围，如果不够平均分配，那么字典序靠前的消费者会被多分配一个分区

假设n=分区数/消费者数量，m=分区数%消费者数量，那么前m个消费者每个分配n+1个分区，后面的（消费者数量-m）个消费者每个分配n个分区

假设消费组内有2个消费者C0和C1，都订阅了主题t0和t1，并且每个主题都有4个分区，那么订阅的所有分区可以标识为：t0p0、t0p1、t0p2、t0p3、t1p0、t1p1、t1p2、t1p3。最终的分配结果为：

消费者C0：t0p0、t0p1、t1p0、t1p1

消费者C1：t0p2、t0p3、t1p2、t1p3

假设2个主题都只有3个分区，那么订阅的所有分区可以标识为：t0p0、t0p1、t0p2、t1p0、t1p1、t1p2。最终的分配结果为：

消费者C0：t0p0、t0p1、t1p0、t1p1

消费者C1：t0p2、t1p2

2）、RoundRobinAssignor分配策略

RoundRobinAssignor分配策略的原理是将消费组内所有消费者及消费者订阅的所有主题的分区按照字典序排序，然后通过轮询方式逐个将分区一次分配给每个向消费者。RoundRobinAssignor分配策略对应的partition.assignment.strategy参数值为org.apache.kafka.clients.consumer.RoundRobinAssignor

如果同一个消费组内所有的消费者的订阅信息都是相同的，那么RoundRobinAssignor分配策略的分区分配会是均匀的

假设消费组内有2个消费者C0和C1，都订阅了主题t0和t1，并且每个主题都有3个分区，那么订阅的所有分区可以表示为：t0p0、t0p1、t0p2、t1p0、t1p1、t1p2。最终分配结果为：

消费者C0：t0p0、t0p2、t1p1

消费者C1：t0p1、t1p0、t1p2

如果同一个消费组内的消费者订阅的信息不是相同的，那么在执行分区分配的时候就不是完全的轮询分配，有可能导致分区分配得不均匀

假设消费组内有3个消费者（C0、C1和C2），它们共订阅了3个主题（t0、t1和t2），这3个主题分别有1、2、3个分区，即整个消费组订阅了t0p0、t1p0、t1p1、t2p0、t2p1、t2p2这6个分区，消费者C0订阅了主题t0，消费者C1订阅的是主题t0和t1，消费者C2订阅的是主题t0、t1和t2，那么最终的分配结果是：

消费者C0：t0p0

消费者C1：t1p0

消费者C2：t1p1、t2p0、t2p1、t2p2

3）、StickyAssignor分配策略

引入这个策略主要有两个目的：

• 分区的分配要尽可能均匀
• 分区的分配尽可能与上次分配的保持相同

当两者发生冲突时，第一个目标优先于第二个目标

假设消费组内有3个消费者（C0、C1和C2），它们都订阅了4个主题（t0、t1、t2、t3），并且每个主题有2个分区。也就是说，整个消费组订阅了t0p0、t0p1、t1p0、t1p1、t2p0、t2p1、t3p0、t3p1这8个分区。最终的分配结果如下：

消费者C0：t0p0、t1p1、t3p0

消费者C1：t0p1、t2p0、t3p1

消费者C2：t1p0、t2p1

和RoundRobinAssignor分配策略所分配的结果相同，假设此时消费者C1脱离了消费组，那么消费组就会执行再均衡操作， 进而消费分区会重新分配。如果采用RoundRobinAssignor分配策略，那么此时的分配结果如下：

消费者C0：t0p0、t1p0、t2p0、t3p0

消费者C2：t0p1、t1p1、t2p1、t3p1

RoundRobinAssignor分配策略会按照消费者C0和C2进行重新轮询分配。如果使用的是StickyAssignor分配策略，分配结果为：

消费者C0：t0p0、t1p1、t3p0、t2p0

消费者C2：t1p0、t2p1、t0p1、t3p1

分配结果中保留了上一次分配中对消费者C0和C2的所有分配结果，并将原来消费者C1的负担分配给了剩余的两个消费者C0和C2，最终C0和C2的分配还保持了均衡

StickyAssignor分配策略尽可能地让前后两次分配相同，进而减少系统资源的损耗及其他异常情况的发生

假设消费组内有3个消费者（C0、C1和C2），它们共订阅了3个主题（t0、t1和t2），这3个主题分别有1、2、3个分区，即整个消费组订阅了t0p0、t1p0、t1p1、t2p0、t2p1、t2p2这6个分区，消费者C0订阅了主题t0，消费者C1订阅的是主题t0和t1，消费者C2订阅的是主题t0、t1和t2，那么最终的分配结果是

消费者C0：t0p0

消费者C1：t1p0、t1p1

消费者C2：t2p0、t2p1、t2p2

4）、自定义分区分配策略

自定义的分配策略必须要实现org.apache.kafka.clients.consumer.internals.PartitionAssignor接口

public interface PartitionAssignor {

    Subscription subscription(Set<String> topics);

    Map<String, Assignment> assign(Cluster metadata, Map<String, Subscription> subscriptions);

    void onAssignment(Assignment assignment);

    String name();

    class Subscription {
        private final List<String> topics;
        private final ByteBuffer userData;

    }

    class Assignment {
        private final List<TopicPartition> partitions;
        private final ByteBuffer userData;

    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

PartitionAssignor接口中定义了两个内部类：Subscription和Assignment

Subscription类用来表示消费者的订阅信息，类中有两个属性：topics和userData，分别表示消费者的订阅主题列表和用户自定义信息

Assignment类用来表示分配结果信息，类中也有两个属性：partitions和userData，分别表示所分配到的分区集合和用户自定义的数据

onAssignment()方法是在每个消费者收到消费组leader分配结果时的回调函数

name()方法用来提供分配策略的名称，自定义的分配策略要注意命名不要与已存在的分配策略发生冲突

真正的分区分配方案的实现是在assign()方法中，方法中的参数metadata表示集群的元数据信息，而subscriptions表示消费组内各个消费者成员的订阅信息，最终方法返回各个消费者的分配信息

Kafka还提供了一个抽象类org.apache.kafka.clients.consumer.internals.AbstractPartitionAssignor，可以简化实现PartitionAssignor接口的工作，并对assign()方法进行了详细实现，其中会将Subscription中的userData信息去掉后再进行分配，Kafka提供的3中分配策略都继承自这个抽象类

2、消费者协调器和组协调器

1）、再均衡的原理

消费者客户端将全部消费者组分成多个子集，每个消费组的子集在服务端对应一个GroupCoordinator对其进行管理，GroupCoordinator是Kafka服务端中用于管理消费组的组件。而消费者客户端中的ConsumerCoordinator组件负责与GroupCoordinator进行交互

ConsumerCoordinator与GroupCoordinator之间最重要的职责就是负责执行消费者再均衡的操作，包括分区分配的工作也是在再均衡期间完成的

当有消费者加入消费组时，消费者、消费组及组协调器之间会经历一下几个阶段

第一阶段（FIND_COORDINATOR）

消费者需要确定它所属的消费组对应的GroupCoordinator所在的broker，并创建与该broker相互通信的网络连接。如果消费者已经保存了与消费组对应的GroupCoordinator节点的信息，并且与它之间的网络连接是正常的，那么就可以进入第二阶段。否则，就需要向集群中的负载最小的节点发送FindCoordinatorRequest请求来查找对应的GroupCoordinator

FindCoordinatorRequest请求体中只有两个域：coordinator_key和coordinator_type。coordinator_key是消费组的名称，即groupId，coordinator_type置为0

Kafka在收到FindCoordinatorRequest请求之后，会根据coordinator_key查找对应的GroupCoordinator节点，找到对应的GroupCoordinator则会返回其相应的nod_id、host和port信息

第二阶段（JOIN_GROUP）

在成功找到消费组所对应的GroupCoordinator之后就进入加入消费组的阶段，在此阶段的消费者会向GroupCoordinator发送JoinGroupRequest请求，并处理响应

JoinGroupRequest的结构包含多个域：

• group_id就是消费组的id
• session_timeout对应消费端参数session.timeout.ms，默认值为10000，即10秒。GroupCoordinator超过session_timeout指定的时间内没有收到心跳报文则认为此消费者已经下线
• rebalance_timeout对应消费端参数max.poll.interval.ms，默认值为300000，即5分钟。表示当下消费组再均衡的时候，GroupCoordinator等待各个消费者重新加入的最长等待时间
• member_id表示GroupCoordinator分配给消费者的id标识。消费者第一次发送JoinGroupRequest请求的时候此字段设置为null
• protocol_type表示消费组实现的协议，对于此消费者而言此字段值为consumer

JoinGroupRequest中的group_protocols域为数组类型，其中可以囊括多个分区分配策略，主要取决于消费者客户端参数partition.assignment.strategy的配置。如果配置了多种策略，那么JoinGroupRequest中就会包含多个protocol_name和protocol_metadata

消费者在发送JoinGroupRequest请求之后会阻塞等待Kafka服务端的响应。服务端在收到JoinGroupRequest请求后会交由GroupCoordinator来进行处理。GroupCoordinator首先会对JoinGroupRequest请求做合法性校验。如果消费者是第一次请求加入消费组，那么JoinGroupRequest请求中的member_id值为null，此时组协调器负责为此消费者生成一个member_id（clientId+"-"+UUID）

1）选举消费组的leader

GroupCoordinator需要为消费组内的消费者选举出一个消费组的leader。如果消费组内还没有leader，那么第一个加入消费组的消费者即为消费组的leader。如果某一时刻leader消费者退出了消费组，那么会重新选举一个新的leader。在GroupCoordinator中消费者的信息是以HashMap的形式存储的，其中key为消费者的member_id，而value是消费者相关的元数据信息。leaderId表示leader消费者的member_id，它的取值为HashMap中的第一个键值对的key

2）选举分区分配策略

分区分配的选举并非由leader消费者决定，而是根据消费组内的各个消费者投票来决定

A.收集各个消费者支持的所有分配策略，组成候选集candidates

B.每个消费者从候选集candidates中找出第一个自身支持的策略，为这个策略投上一票

C.计算候选集中各个策略的选票数，选票数最多的策略即为当前消费组的分配策略

消费者所支持的分配策略是partition.assignment.strategy参数配置的策略

在此之后，Kafka服务端就要发送JoinGroupResponse响应给各个消费者，JoinGroupResponse中包含GroupCoordinator中投票选举出的分配策略的信息。并且，只有leader消费者的JoinGroupResponse中包含各个消费者的订阅信息。Kafka把分区策略的具体分配交还给客户端，自身并不参与具体的分配细节，这样即使以后分区分配的策略发生了变更，只需要重启消费端的应用即可，而不需要重启服务端

客户端发送JoinGroupRequest请求：

客户端接收JoinGroupResponse响应：

第三阶段（SYNC_GROUP）

在第三个阶段，也就是同步阶段，各个消费者GroupCoordinator发送SyncGroupRequest请求来同步分配方案

GroupCoordinator会先对SyncGroupRequest请求做合法性校验，在此之后会将从leader消费者发送过来的分配方案提取出来，连同整个消费组的元数据信息一起存入Kafka的__consumer_offsets主题中，最后发送SyncGroupResponse给各个消费者各自所属的分配方案

当消费者收到所属的分配方案之后会调用PartitionAssignor中的onAssignment()方法。随后再调用ConsumerRebalanceListener中的onPartitionsAssigned()方法。之后开启心跳任务，消费者定期向服务端的GroupCoordinator发送HeartBeatRequest来确定彼此在线

第四阶段（HEARTBEAT）

消费组中的所有消费者处于正常工作状态。在正式消费之前，消费者还需要确定拉取消息的起始位置。假设之前已经将最后的消费位移提交到了GroupCoordinator，并且GroupCoordinator将其保存到了Kafka内部的__consumer_offsets主题中，此时消费者可以通过OffsetFetchRequest请求获取上次提交的消费位移并从此处继续消费

消费者通过向GroupCoordinator发送心跳来维持它们与消费组的从属关系，以及它们对分区的所有权关系。只要消费者以正常的时间间隔发送心跳，就被认为是活跃的，说明它还在读取分区中的消息。心跳线程是一个独立的线程，可以在轮询消息的空档发送心跳。如果消费者停止发送心跳的时间足够长，则整个会话被判定为过期，GroupCoordinator也会认为这个消费者已经死亡，就会触发一次再均衡行为

3、事务

1）、消息传输保障

消息中间件的消息传输保障有3个层级：

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

2）、幂等

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

开启幂等性功能需要显示地将生产者客户端参数enable.idempotence设置为true即可（这个参数默认值为false）

        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
1

如果要确保幂等性功能正常，需要确保生产者客户端的retries、acks、max.in.flight.requests.per.connection这几个参数不被配置错。在使用幂等性功能的时候，用户完全不需要配置这几个参数

如果用户显示指定了retries参数，那么这个参数的值必须大于0，如果没有显示指定retries参数，那么KafkaProducer会将它置为Integer.MAX_VALUE。同时还要保证max.in.flight.requests.per.connection（限制每个连接最多缓存的请求数）参数的值不能大于5，acks参数的值为-1

为了实现生产者的幂等性，Kafka为此引入了producer id（PID）和序列号这两个概念。每个新的生产者实例在初始化的时候都会被分配一个PID，这个PID对用户而言是完全透明的。对于每个PID，消息发送到的每一个分区都有对应的序列号，这些序列号从0开始单调递增。生产者每发送一条消息就会将<PID，分区>对应的序列号的值加1

broker端会在内存中为每一对<PID，分区>维护一个序列号。对于收到的每一条消息，只有当它的序列号的值比broker端中维护的对应的序列号的值大1（即SN_new=SN_old+1）时，broker才会接收它。如果SN_new<SN_old+1，那么说明消息被重复写入，broker可以直接将其丢弃。如果SN_new>SN_old+1，那么说明中间有数据尚未写入，出现了乱序，可能有消息丢失，对应的生产者会抛出OutOfOrderSequenceException

引入序列号来实现幂等性只是针对每一对<PID，分区>而言的，也就是说，Kafka的幂等性只能保证单个生产者会话中单分区的幂等

        ProducerRecord<String, String> record = new ProducerRecord<>(topic, "key", "msg");
        producer.send(record);
        producer.send(record);
1
2
3

上面示例中发送了两条相同的消息，不过这仅仅是指消息内容相同，但对Kafka而言是两条不同的消息，因为会为这两条消息分配不同的序列号。Kafka并不会保证消息内容的幂等

3）、事务

事务可以保证对多个分区写入操作的原子性。Kafka中的事务可以使应用程序将消费消息、生产消息、提交消费位移当作原子操作来处理，同时成功或失败，即使该生产或消费会跨多个分区

为了实现事务，应用程序必须提供唯一的transactionId，这个transactionId通过客户端参数transactional.id来显示指定

        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactionId");
1

事务要求生产者开启幂等特性，因此通过将transactional.id参数设置为非空从而开启事务特性的同时需要将enable.idempotence设置为true（如果未显示指定，则KafkaProducer默认会将它的值设置为true）

transactionId与PID一一对应，两者之间所不同的是transactionId由用户显示指定，而PID是由Kafka内部分配的。另外，为了保证新的生产者启动后具有相同transactionId的旧生产者能够立即失效，每个生产者通过transactionId获取PID的同时，还会获取一个单调递增的producer epoch。如果使用同一个transactionId开启两个生产者，那么前一个开启的生产者会报错

从生产者的角度分析，通过事务，Kafka可以保证跨生产者会话的消息幂等发送，以及跨生产者会话的事务恢复。前者表示具有相同transactionId的新生产者实例被创建且工作的时候，旧的且拥有相同transactionId的生产者实例将不再工作。后者指当某个生产者实例宕机后，新的生产者实例可以保证任何未完成的旧事物要么被提交，要么被中止，如此可以是新的生产者实例从一个正常的状态开始工作

从消费者的角度分析，Kafka并不能保证已提交的事务中的所有消息都能够被消费：

• 对采用日志压缩策略的主题而言，事务中的某些消息有可能被清理（相同key的消息，后写入的消息会覆盖前面写入的消息）
• 事务中消息可能分布在同一个分区的多个日志分段中，当老的日志分段被删除时，对应的消息可能会丢失
• 消费者可以通过seek()方法访问任意offset的消息，从而可能遗漏事务中的部分消息
• 消费者在消费时可能没有分配到事务内的所有分区，如此它也就不能读取事务中的所有消息

KafkaProducer提供了5个与事务相关的方法

//初始化事务
void initTransactions()
//开启事务
void beginTransaction() throws ProducerFencedException
//消费者在事务内的位移提交
void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
							  String consumerGroupId) throws ProducerFencedException
//提交事务							  
void commitTransaction() throws ProducerFencedException
//中止事务
void abortTransaction() throws ProducerFencedException
1
2
3
4
5
6
7
8
9
10
11

事务消息发送示例：

        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionId);
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        producer.initTransactions();
        producer.beginTransaction();
        try {
            ProducerRecord<String, String> record1 = new ProducerRecord<>(topic, "msg1");
            producer.send(record1);
            ProducerRecord<String, String> record2 = new ProducerRecord<>(topic, "msg2");
            producer.send(record2);
            producer.commitTransaction();
        } catch (ProducerFencedException e) {
            producer.abortTransaction();
        } finally {
            producer.close();
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

在消费端有一个参数isolation.level，默认值为read_uncommitted，意思是说消费端应用可以消费到未提交的事务。还可以设置为read_committed，表示消费端应用只能看到提交的事务

日志文件中除了普通的消息，还有一种消息专门用来标志一个事务的结束，它就是控制消息（ControlBatch）。控制消息一共有两种类型：COMMIT和ABORT，分别用来表征事务已经成功提交或已经被成功中止。KafkaConsumer可以通过这个控制消息来判断对应的事务是被提交了还是被中止了，然后结合参数isolation.level配置的隔离级别来决定是否将相应的消息返回给消费端应用

consume-transform-produce（消费-转换-生产）示例：

@RunWith(SpringRunner.class)
@SpringBootTest
public class TransactionConsumerTransformProduceTest {
    public static final String brokerList = "192.168.126.158:9092";

    public Properties getConsumerProperties() {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "groupId");
        return props;
    }

    public Properties getProducerProperties() {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactionId");
        return props;
    }

    @Test
    public void consumerTransformProduceTest() {
        //初始化生产者和消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(getConsumerProperties());
        consumer.subscribe(Collections.singleton("topic-source"));
        KafkaProducer<String, String> producer = new KafkaProducer<>(getProducerProperties());
        //初始化事务
        producer.initTransactions();
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            if (!records.isEmpty()) {
                Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
                //开启事务
                producer.beginTransaction();
                try {
                    for (TopicPartition partition : records.partitions()) {
                        List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
                        for (ConsumerRecord<String, String> record : partitionRecords) {
                            System.out.println("topic=" + record.topic() + ",partition=" + record.partition() + ",offset=" + record.offset());
                            System.out.println("key=" + record.key() + ",value=" + record.value());
                            ProducerRecord<String, String> producerRecord = new ProducerRecord<>("topic-sink", record.key(), record.value());
                            //消费-生产模型
                            producer.send(producerRecord);
                        }
                        long lastConsumedOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
                        offsets.put(partition, new OffsetAndMetadata(lastConsumedOffset + 1));
                    }
                    //提交消费位移
                    producer.sendOffsetsToTransaction(offsets, "groupId");
                    //提交事务
                    producer.commitTransaction();
                } catch (ProducerFencedException e) {
                    e.printStackTrace();
                    producer.abortTransaction();
                }
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

为了实现事务的功能，Kafka引入了事务协调器（TransactionCoordinator）。每一个生产者都会被指派一个特定的TransactionCoordinator，所有的事务逻辑包括分派PID等都是由TransactionCoordinator来负责实施的。TransactionCoordinator会将事务状态持久化到内部主题__transaction_state中

以consume-transform-produce的流程为例分析Kafka事务的实现原理

1）查找TransactionCoordinator

生产者向Kafka发送FindCoordinatorRequest请求，Kafka在收到FindCoordinatorRequest请求之后，会根据coordinator_key（transactionId）查找对应的TransactionCoordinator节点。如果找到，则会返回其相应的node_id、host和port信息。具体查找TransactionCoordinator的方式是根据transactionId的哈希值计算主题__transaction_state中的分区编号，找到对应的分区之后，再寻找此分区leader副本所在的broker节点，该broker节点即为transactionId对应的TransactionCoordinator节点

2）获取PID

在找到TransactionCoordinator节点之后，需要为当前生产者分配一个PID。凡是开启了幂等性功能的生产者都必须执行这个操作，不需要考虑该生产者是否还开启了事务。生产者获取PID的操作是通过InitProducerIdRequest请求来实现的，InitProducerIdRequest请求体结构如上图所示，其中transactional_id表示事务的transactionId，transaction_timeout_ms表示TransactionCoordinator等待事务状态更新的超时时间，通过生产者客户端参数transaction.timeout.ms配置，默认值为60000

保存PID

生产者的InitProducerIdRequest请求会被发送给TransactionCoordinator。如果没开启事务特性而只开启幂等特性，那么InitProducerIdRequest请求可以发送给任意的broker。当TransactionCoordinator第一次收到包含该transactionId的InitProducerIdRequest请求时，它会把transactionId和对应的PID以消息（事务日志消息）的形式保存到主题__transaction_state中，如上图所示。这样可以保证<transactionId,PID>的对应关系被持久化，从而保证即使TransactionCoordinator宕机该对应关系也不会丢失

其中transaction_status包含Empty(0)、Ongoing（1）、PrepareCommit(2)、PrepareAbort(3)、CompleteCommit(4)、CompleteAbort(5)、Dead(6)这几种状态。在存入主题__transaction_state之前，事务日志消息同样会根据单独的transactionId来计算要发送的分区

InitProducerIdRequest对应的InitProducerIdResponse响应体结构如上图所示，除了返回PID，还会触发执行以下任务：

• 增加该PID对应的producer_epoch。具有相同PID但producer_epoch小于该producer_epoch的其他生产者新开启的事务将被拒绝
• 恢复和终止之前的生产者未完成的事务

3）开启事务

通过KafkaProducer的beginTransaction()方法可以开启一个事务，调用该方法后，生产者本地会标记已经开启一个新的事务，只有在生产者发送第一条消息之后TransactionCoordinator才会认为该事务已经开启

4）Consume-Transform-Produce

这个阶段囊括了整个事务的数据处理过程

AddPartitionsToTxnRequest

当生产者给一个新的分区（TopicPartition）发送数据前，它需要先向TransactionCoordinator发送AddPartitionsToTxnRequest请求，这个请求会让TransactionCoordinator将<transactionId,TopicPartition>的对应关系存储在主题__transaction_state中，有了这个对照关系在后续的步骤中为每个分区设置COMMIT或ABORT标记

如果该分区是对应事务中的第一个分区，那么此时TransactionCoordinator还会启动对该事务的计时

ProduceRequest

生产者通过ProduceRequest请求发送消息（ProducerBatch）到用户自定义主题中，和普通消息不同的是ProducerBatch中会包含实质的PID、producer_epoch和sequence number

AddOffsetsToTxnRequest

通过KafkaProducer的sendOffsetsToTransaction()方法可以在一个事务批次里处理消息的消费和发送，方法中包含2个参数：Map<TopicPartition, OffsetAndMetadata> offsets和String consumerGroupId。这个方法会向TransactionCoordinator节点发送AddOffsetsToTxnRequest请求。TransactionCoordinator收到这个请求之后会通过groupId来推导出在__consumer_offsets中的分区，之后TransactionCoordinator会将整个分区保存在__transaction_state中

TxnOffsetCommitRequest

这个请求也是sendOffsetsToTransaction()方法中的一部分，在处理完AddOffsetsToTxnRequest之后，生产者还会发送TxnOffsetCommitRequest请求给GroupCoordinator，从而将本次事务中包含的消费位移信息offsets存储到主题__consumer_offsets中

5）提交或者中止事务

EndTxnRequest

无论调用commitTransaction()方法还是abortTransaction()方法，生产者都会向TransactionCoordinator发送EndTxnRequest请求，以此来通知它提交事务还是中止事务

TransactionCoordinator在收到EndTxnRequest请求之后会执行如下操作：

A.将PREPARE_COMMIT或PREPARE_ABORT消息写入主题__transaction_state

B.通过WriteTxnMarkersRequest请求将COMMIT或ABORT信息写入用户所使用的普通主题和__consumer_offsets

C.将COMPLETE_COMMIT或COMPLETE_ABORT信息写入内部主题__transaction_state

WriteTxnMarkersRequest

WriteTxnMarkersRequest请求是由TransactionCoordinator发向事务中各个分区的leader节点的，当节点收到这个请求之后，会在相应的分区中写入控制消息（ControlBatch）。控制消息用来标识事务的终结，它和普通的消息一样存储在日志文件中。RecordBatch中attributes字段的第6位用来标识当前消息是否是控制消息

attributes字段中的第5位用来标识当前消息是否处于事务中

控制消息的key和value内部的version值都为0，key中的type表示控制类型：0表示ABORT，1表示COMMIT；value中的coordinator_epoch表示TransactionCoordinator的纪元，TransactionCoordinator切换的时候会更新其值

写入最终的COMPLETE_COMMIT或COMPLETE_ABORT

TransactionCoordinator将最终的COMPLETE_COMMIT或COMPLETE_ABORT信息写入主题__transaction_state以表明当前事务已经结束，此时可以删除主题__transaction_state中所有关于该事务的消息。由于主题__transaction_state采用的日志清理策略为日志压缩，所以这里的删除只需要将相应的消息设置为墓碑消息即可