Kafka -- 控制器

控制器

控制器（Controller）是Kafka的核心组件，主要作用是在ZK的帮助下管理和协调整个Kafka集群
集群中任一Broker都能充当控制器的角色，但在运行过程中，只能有一个Broker成为控制器，行使管理和协调的职责

[zk: localhost:2181(CONNECTED) 1] get /controller
{"version":1,"brokerid":0,"timestamp":"1571311742367"}
cZxid = 0xd68
ctime = Thu Oct 17 19:29:02 CST 2019
mZxid = 0xd68
mtime = Thu Oct 17 19:29:02 CST 2019
pZxid = 0xd68
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x1000209974b0000
dataLength = 54
numChildren = 0

Zookeeper

Kafka控制器重度依赖ZK
ZK是一个提供高可靠性的分布式协调服务框架
ZK使用类似于文件系统的树形结构，根目录以/开始，结构上的每个节点称为znode，用来保存一些元数据协调信息
如果以znode的持久性来划分，znode可以分为持久性znode和临时znode
- 持久性znode不会因为ZK集群重启而消失
- 临时znode则会与创建该znode的ZK会话绑定，一旦会话结束，该节点会被自动删除
ZK赋予客户端监控znode变更的能力，即所谓的Watch通知功能
- 一旦znode节点被创建、删除、子节点数量发生变化，znode所存的数据本身发生变更
- ZK会通过节点变更监听器（ChangeHandler）的方式显式通知客户端
ZK被用来实现集群成员管理、分布式锁、领导者选举等功能，Kafka控制器大量使用Watch功能实现对集群的协调管理

kafka在ZK中创建的znode

`/controller`节点

Broker在启动时，会尝试去ZK创建/controller节点
第一个成功创建/controller节点的Broker会被指定为为控制器

控制器的职责

主题管理
- 完成对Kafka主题的创建、删除以及分区增加的操作
- 执行kafka-topics时，大部分的后台工作都是由控制器完成的
分区重分配
- 分区重分配主要是指kafka-reassign-partitions脚本提供的对已有主题分区进行细粒度的分配功能
Preferred领导者选举
- Preferred领导者选举主要是Kafka为了避免部分Broker负载过重而提供的一种换Leader的方案
集群成员管理
- 自动检测新增Broker、Broker主动关闭、Broker宕机
- 自动检测依赖于Watch功能和ZK临时节点组合实现的
  - 控制器会利用Watch机制检查ZK的/brokers/ids节点下的子节点数量变更
  - 当有新Broker启动后，它会在/brokers/ids/下创建专属的临时znode节点
    - 一旦创建完毕，ZK会通过Watch机制将消息通知推送给控制器，控制器能够自动感知这个变化
  - 当Broker宕机或者主动关闭后，该Broker与ZK的会话结束，这个znode会被自动删除
    - ZK的Watch机制会将这一变更推送给控制器
数据服务
- 向其它Broker提供数据服务，控制器上保存了最全的集群元数据信息
- 其它Broker会定期接收控制器发来的元数据更新请求，从而更新其内存中的缓存数据

控制器保存的数据

所有主题信息，包括具体的分区信息、比如领导者副本是谁，ISR集合有哪些副本
所有Broker信息，包括哪些运行中的Broker，哪些正在关闭的Broker
所有涉及运维任务的分区，包括当前正在进行Preferred领导者选举以及分区重分配的分区列表
上述这些数据在ZK中也保存了一份，每当控制器初始化时，都会从ZK上读取对应的元数据并填充到自己的缓存中
- 有了这些数据，控制器就能对其它Broker提供数据服务了
- 控制器通过向其它Broker发送请求的方式将这些数据同步到其它Broker上

控制器故障转移

故障转移
- 当运行中的控制器突然宕机或者意外终止时，Kafka能快速感知并立即启用备用控制器来代替之前失败的控制器
- 该过程称为FailOver，该过程是自动完成的
一开始，Broker 0是控制器，当Broker 0宕机后，ZK通过Watch机制感知到并删除了/controller临时节点
然后，所有存活的Broker开始竞选新的控制器，Broker 3最终赢得了选举，成功地在ZK上重建了/controller临时节点
之后，Broker 3会从ZK中读取集群元数据信息，并初始化到自己的缓存中，至此控制器的FailOver完成

控制器内部设计原理

在Kafka 0.11之前，控制器的设计是相当繁琐的，导致很多Bug无法修复
控制器是多线程的设计，会在内部创建很多个线程
- 控制器需要为每个Broker都创建一个对应的Socket连接，然后再创建一个专属的线程，用于向这些Broker发送请求
- 控制器连接ZK的会话，也会创建单独的线程来处理Watch机制的通知回调
- 控制器还会为主题删除创建额外的IO线程
这些线程还会访问共享的控制器缓存数据，多线程访问共享可变数据是维持线程安全的最大难题
- 为了保护数据安全性，控制器在代码中大量使用ReentrantLock同步机制，进一步拖慢整个控制器的处理速度
社区在0.11版本重构了控制器的底层设计，把多线程的方案改成了单线程+事件队列的方案
- 引进了事件处理器，统一处理各种控制器事件
- 控制器将原来执行的操作全部建模成独立的事件，发送到专属的事件队列中，供事件处理器消费
- 单线程：控制器只是把缓存状态变更方面的工作委托给了这个线程而已
- 优点：控制器缓存中保存的状态只被一个线程处理，因此不需要重量级的线程同步机制来维护线程安全
针对控制器的第二个改进：将之前同步操作ZK全部换成异步操作
- ZK本身的API提供了同步写和异步写两种方式
- 之前控制器操作ZK时使用的是同步API，性能很差
  - 当有大量主题分区发生变更时，ZK容易成为系统的瓶颈
Kafka从2.2开始，将控制器发送的请求和普通数据类的请求分开，实现控制器请求单独处理的逻辑
- 之前Broker对接收到的所有请求都一视同仁，不会区别对待
- 如果删除了某个主题，那么控制器会给主题的所有副本所在的Broker发送StopReplica请求
- 如果此时Broker上有大量Produce请求堆积，那么StopReplica请求只能排队
- 既然主题都要被删除了，继续处理Produce请求就显得很没有意义