Kafka核心组件之控制器和协调器

[TOC]

我们已经知道Kafka的集群由n个的broker所组成，每个broker就是一个kafka的实例或者称之为kafka的服务。其实控制器也是一个broker，控制器也叫leader broker。

他除了具有一般broker的功能外，还负责分区leader的选取，也就是负责选举partition的leader replica。

kafka每个broker启动的时候，都会实例化一个KafkaController，并将broker的id注册到zookeeper，集群在启动过程中，通过选举机制选举出其中一个broker作为leader，也就是前面所说的控制器。

包括集群启动在内，有三种情况触发控制器选举：

1、集群启动

2、控制器所在代理发生故障

3、zookeeper心跳感知，控制器与自己的session过期

按照惯例，先看图。我们根据下图来讲解集群启动时，控制器选举过程。

假设此集群有三个broker，同时启动。

（一）3个broker从zookeeper获取/controller临时节点信息。/controller存储的是选举出来的leader信息。此举是为了确认是否已经存在leader。

（二）如果还没有选举出leader，那么此节点是不存在的，返回-1。如果返回的不是-1，而是leader的json数据，那么说明已经有leader存在，选举结束。

（三）三个broker发现返回-1，了解到目前没有leader，于是均会触发向临时节点/controller写入自己的信息。最先写入的就会成为leader。

（四）假设broker 0的速度最快，他先写入了/controller节点，那么他就成为了leader。而broker1、broker2很不幸，因为晚了一步，他们在写/controller的过程中会抛出ZkNodeExistsException，也就是zk告诉他们，此节点已经存在了。

经过以上四步，broker 0成功写入/controller节点，其它broker写入失败了，所以broker 0成功当选leader。

此外zk中还有controller_epoch节点，存储了leader的变更次数，初始值为0，以后leader每变一次，该值+1。所有向控制器发起的请求，都会携带此值。如果控制器和自己内存中比较，请求值小，说明kafka集群已经发生了新的选举，此请求过期，此请求无效。如果请求值大于控制器内存的值，说明已经有新的控制器当选了，自己已经退位，请求无效。kafka通过controller_epoch保证集群控制器的唯一性及 *** 作的一致性。

由此可见，Kafka控制器选举就是看谁先争抢到/controller节点写入自身信息。

控制器的初始化，其实是初始化控制器所用到的组件及监听器，准备元数据。

前面提到过每个broker都会实例化并启动一个KafkaController。KafkaController和他的组件关系，以及各个组件的介绍如下图：

图中箭头为组件层级关系，组件下面还会再初始化其他组件。可见控制器内部还是有些复杂的，主要有以下组件：

1、ControllerContext，此对象存储了控制器工作需要的所有上下文信息，包括存活的代理、所有主题及分区分配方案、每个分区的AR、leader、ISR等信息。

2、一系列的listener，通过对zookeeper的监听，触发相应的 *** 作，黄色的框的均为listener

3、分区和副本状态机，管理分区和副本。

4、当前代理选举器ZookeeperLeaderElector，此选举器有上位和退位的相关回调方法。

5、分区leader选举器，PartitionLeaderSelector

6、主题删除管理器，TopicDeletetionManager

7、leader向broker批量通信的ControllerBrokerRequestBatch。缓存状态机处理后产生的request，然后统一发送出去。

8、控制器平衡 *** 作的KafkaScheduler，仅在broker作为leader时有效。

Kafka集群的一些重要信息都记录在ZK中，比如集群的所有代理节点、主题的所有分区、分区的副本信息(副本集、主副本、同步的副本集)。每个broker都有一个控制器，为了管理整个集群Kafka选利用zk选举模式，为整个集群选举一个“中央控制器”或”主控制器“，控制器其实就是一个broker节点，除了一般broker功能外，还具有分区首领选举功能。中央控制器管理所有节点的信息，并通过向ZK注册各种监听事件来管理整个集群节点、分区的leader的选举、再平衡等问题。外部事件会更新ZK的数据，ZK中的数据一旦发生变化，控制器都要做不同的响应处理。

故障转移其实就是leader所在broker发生故障，leader转移为其他的broker。转移的过程就是重新选举leader的过程。

重新选举leader后，需要为该broker注册相应权限，调用的是ZookeeperLeaderElector的onControllerFailover()方法。在这个方法中初始化和启动了一系列的组件来完成leader的各种 *** 作。具体如下，其实和控制器初始化有很大的相似度。

1、注册分区管理的相关监听器

2、注册主题管理的相关监听

3、注册代理变化监听器

4、重新初始化ControllerContext，

5、启动控制器和其他代理之间通信的ControllerChannelManager

6、创建用于删除主题的TopicDeletionManager对象,并启动。

7、启动分区状态机和副本状态机

8、轮询每个主题，添加监听分区变化的PartitionModificationsListener

9、如果设置了分区平衡定时 *** 作，那么创建分区平衡的定时任务，默认300秒检查并执行。

除了这些组件的启动外，onControllerFailover方法中还做了如下 *** 作：

1、/controller_epoch值+1，并且更新到ControllerContext

2、检查是否出发分区重分配，并做相关 *** 作

3、检查需要将优先副本选为leader，并做相关 *** 作

4、向kafka集群所有代理发送更新元数据的请求。

下面来看leader权限被取消时，调用的方法onControllerResignation

1、该方法中注销了控制器的权限。取消在zookeeper中对于分区、副本感知的相应监听器的监听。

2、关闭启动的各个组件

3、最后把ControllerContext中记录控制器版本的数值清零，并设置当前broker为RunnignAsBroker，变为普通的broker。

通过对控制器启动过程的学习，我们应该已经对kafka工作的原理有了了解， 核心是监听zookeeper的相关节点，节点变化时触发相应的 *** 作 。

有新的broker加入集群时，称为代理上线。反之，当broker关闭，推出集群时，称为代理下线。

代理上线：

1、新代理启动时向/brokers/ids写数据

2、BrokerChangeListener监听到变化。对新上线节点调用controllerChannelManager.addBroker()，完成新上线代理网络层初始化

3、调用KafkaController.onBrokerStartup()处理

3.5恢复因新代理上线暂停的删除主题 *** 作线程

代理下线：

1、查找下线节点集合

2、轮询下线节点，调用controllerChannelManager.removeBroker()，关闭每个下线节点网络连接。清空下线节点消息队列，关闭下线节点request请求

3、轮询下线节点，调用KafkaController.onBrokerFailure处理

4、向集群全部存活代理发送updateMetadataRequest请求

顾名思义，协调器负责协调工作。本节所讲的协调器，是用来协调消费者工作分配的。简单点说，就是消费者启动后，到可以正常消费前，这个阶段的初始化工作。消费者能够正常运转起来，全有赖于协调器。

主要的协调器有如下两个：

1、消费者协调器（ConsumerCoordinator）

2、组协调器（GroupCoordinator）

kafka引入协调器有其历史过程，原来consumer信息依赖于zookeeper存储，当代理或消费者发生变化时，引发消费者平衡，此时消费者之间是互不透明的，每个消费者和zookeeper单独通信，容易造成羊群效应和脑裂问题。

为了解决这些问题，kafka引入了协调器。服务端引入组协调器（GroupCoordinator），消费者端引入消费者协调器（ConsumerCoordinator）。每个broker启动的时候，都会创建GroupCoordinator实例，管理部分消费组（集群负载均衡）和组下每个消费者消费的偏移量（offset）。每个consumer实例化时，同时实例化一个ConsumerCoordinator对象，负责同一个消费组下各个消费者和服务端组协调器之前的通信。如下图：

消费者协调器，可以看作是消费者做 *** 作的代理类（其实并不是），消费者很多 *** 作通过消费者协调器进行处理。

消费者协调器主要负责如下工作：

1、更新消费者缓存的MetaData

2、向组协调器申请加入组

3、消费者加入组后的相应处理

4、请求离开消费组

5、向组协调器提交偏移量

6、通过心跳，保持组协调器的连接感知。

7、被组协调器选为leader的消费者的协调器，负责消费者分区分配。分配结果发送给组协调器。

8、非leader的消费者，通过消费者协调器和组协调器同步分配结果。

消费者协调器主要依赖的组件和说明见下图：

可以看到这些组件和消费者协调器担负的工作是可以对照上的。

组协调器负责处理消费者协调器发过来的各种请求。它主要提供如下功能：

组协调器在broker启动的时候实例化，每个组协调器负责一部分消费组的管理。它主要依赖的组件见下图：

这些组件也是和组协调器的功能能够对应上的。具体内容不在详述。

下图展示了消费者启动选取leader、入组的过程。

消费者入组的过程，很好的展示了消费者协调器和组协调器之间是如何配合工作的。leader consumer会承担分区分配的工作，这样kafka集群的压力会小很多。同组的consumer通过组协调器保持同步。消费者和分区的对应关系持久化在kafka内部主题。

消费者消费时，会在本地维护消费到的位置（offset），就是偏移量，这样下次消费才知道从哪里开始消费。如果整个环境没有变化，这样做就足够了。但一旦消费者平衡 *** 作或者分区变化后，消费者不再对应原来的分区，而每个消费者的offset也没有同步到服务器，这样就无法接着前任的工作继续进行了。

因此只有把消费偏移量定期发送到服务器，由GroupCoordinator集中式管理，分区重分配后，各个消费者从GroupCoordinator读取自己对应分区的offset，在新的分区上继续前任的工作。

下图展示了不提交offset到服务端的问题：

开始时，consumer 0消费partition 0 和1，后来由于新的consumer 2入组，分区重新进行了分配。consumer 0不再消费partition2，而由consumer 2来消费partition 2，但由于consumer之间是不能通讯的，所有consumer2并不知道从哪里开始自己的消费。

因此consumer需要定期提交自己消费的offset到服务端，这样在重分区 *** 作后，每个consumer都能在服务端查到分配给自己的partition所消费到的offset，继续消费。

由于kafka有高可用和横向扩展的特性，当有新的分区出现或者新的消费入组后，需要重新分配消费者对应的分区，所以如果偏移量提交的有问题，会重复消费或者丢消息。偏移量提交的时机和方式要格外注意！！

1、自动提交偏移量

设置 enable.auto.commit为true，设定好周期，默认5s。消费者每次调用轮询消息的poll() 方法时，会检查是否超过了5s没有提交偏移量，如果是，提交上一次轮询返回的偏移量。

这样做很方便，但是会带来重复消费的问题。假如最近一次偏移量提交3s后，触发了再均衡，服务器端存储的还是上次提交的偏移量，那么再均衡结束后，新的消费者会从最后一次提交的偏移量开始拉取消息，此3s内消费的消息会被重复消费。

2、手动提交偏移量

设置 enable.auto.commit为false。程序中手动调用commitSync()提交偏移量，此时提交的是poll方法返回的最新的偏移量。

commitSync()是同步提交偏移量，主程序会一直阻塞，偏移量提交成功后才往下运行。这样会限制程序的吞吐量。如果降低提交频次，又很容易发生重复消费。

这里我们可以使用commitAsync()异步提交偏移量。只管提交，而不会等待broker返回提交结果

commitSync只要没有发生不可恢复错误，会进行重试，直到成功。而commitAsync不会进行重试，失败就是失败了。commitAsync不重试，是因为重试提交时，可能已经有其它更大偏移量已经提交成功了，如果此时重试提交成功，那么更小的偏移量会覆盖大的偏移量。那么如果此时发生再均衡，新的消费者将会重复消费消息。

kafka producer将消息发送给broker后，消息日志会被存储在broker的磁盘上，采用顺序写入的方式。顺序写可以加快磁盘访问速度，并且可以将将多个小型的逻辑写合并成一次大型的物理磁盘写入，官方数据显示顺序写比随机写入快6000倍以上。另外， *** 作系统使用内存对磁盘进行缓存即pagecache，pagecache完全由 *** 作系统管理，这也使得写数据变得即简洁也快速。

配置中可以调整过期时间，超过改时间的消息日志将移除，默认值为7天；也可配置文件大小阈值，达到该阈值后，从最旧消息开始删除。配置项为：

从文件到套接字的常见数据传输路径有4步：

1). *** 作系统从磁盘读取数据到内核空间的 pagecache

2).应用程序读取内核空间的数据到用户空间的缓冲区

3).应用程序将数据(用户空间的缓冲区)写回内核空间到套接字缓冲区(内核空间)

4). *** 作系统将数据从套接字缓冲区(内核空间)复制到通过网络发送的 NIC 缓冲区

kafka使用 producer ，broker 和 consumer 都共享的标准化的二进制消息格式，这样数据块不用修改就能在他们之间传递。kafka采用Linux 中系统调用sendfile的方式，直接将数据从 pagecache 转移到 socket 网络连接中。这种零拷贝方式使得kafka数据传输更加高效。

以前面文章中安装的kafka为例： Mac 安装kafka

kafka本地文件存储目录可以在配置文件server.properties中设置，参数及默认值为：

进入该目录，可以看到kafka保存的cosumer offset和topic消息：

其中__consumer_offsets开头的为消费的offset信息，test1开头的即为之前创建的topic “test1”，该topic有三个分区，分区编号从0开始，分别是test1-0、test1-1、test1-2。

进入test1-0，查看包含文件如下：

可以看到kafka消息按partition存储的，每个partition一个目录。partition下消息分段(segment)存储，默认每段最大1G，通过参数log.segment.bytes可配置。segment包含索引文件index、消息文件log，分别存储消息的索引和内容，以.index和.log结尾，文件命名为当前segment第一个消息offset。index文件在log每隔一定数据量之间建立索引，可以通过参数index.interval.bytes配置。

通过kafka命令查看00000000000000000000.index内容如下：

00000000000000000000.log内容如下：

其中索引文件中包含两个字段：（offset，position），分别表示消息offset和该消息在log文件的偏移量。如上图中offset=0的消息对应的position=0；对应的就是00000000000000000000.log中的第一条消息：

其中payload为具体的消息内容。

另外里面还有一个以".timeindex"结尾的文件，查看其内容：

该日志文件是kafka0.10.1.1加入的，其中保存的为：（消息时间戳，offset）。时间戳是该segment最后一个消息对应的时间戳（与log文件中最后一条记录时间戳一致），kafka也支持根据时间来读取消息。

由上可知消息是按partition来存储的，partition可以配置n个副本followers。多个partition和其follower在broker上是怎么分配的呢？

partition和broker都进行了排序，下标从0开始；

假设有k个broker，第i个partition被分配到到 i%k 个broker上；

第i%k个broker即为partition i 的leader，负责这个partition的读写 ；

partition的followers也进行排序，从leader的后续broker开始分配，第i个partition的第j个副本broker为 （j+ i%k）%k。

一个有3个broker的kafka集群，包含3个partition，每个partition副本数为1的topic如下图：

总结：

kafka将消息日志采用顺序写入的方式存放在broker磁盘中；数据传输通过系统调用sendfile零拷贝方式；消息日志分段存放，可配置清除时间或大小阈值；每段包含消息索引、消息内容两个文件，通过索引实现快速查找；按照/topic/partition的目录结构分开存储，且均匀分布到集群各broker上。

参考：

https://kafka.apachecn.org/documentation.html#design

软件准备

Confluent平台是一个可靠的，高性能的流处理平台，你可以通过这个平台组织和管理各式各样的数据源中的数据。

confluent start 会启动 confluent 全部组件，如果想要单独启动，比如单独启动 schema-registry，可以执行以下命令：

具体的单独启动各组件的命令，进入 bin 目录下，一看就能明白，不再赘述。

说明：

confluent 中内嵌了 Kafka 和 Zookeeper，你也可以通过指定不同的 zookeeper 在其他的 kafka 集群中创建 topic 或执行其他 *** 作。

以上命令是内嵌的一个kafka-producer脚本，生成随机的用户信息，可以通过 quickstart=[CLICKSTREAM_CODES, CLICKSTREAM, CLICKSTREAM_USERS, ORDERS, RATINGS, USERS, USERS_, PAGEVIEWS] 来生成不同的数据，这个脚本会运行很长时间(官网只说了很长时间，到底多长，没说)，除非你手动停止

在另一个窗口中，进入KSQL命令行(上一个窗口继续发数据不要停)

把生产过来的数据创建为user表：

设置消费偏移量为 "earliest"：

查询：

---