你必定不能错过的Kafka控制器

控制器组件（Controller），是 Apache Kafka 的核心组件。它的主要做用是在 ApacheZooKeeper 的帮助下管理和协调整个 Kafka 集群。集群中任意一台 Broker 都能充当控制器的角色，可是，在运行过程当中，只能有一个 Broker 成为控制器，行使其管理和协调的职责。换句话说，每一个正常运转的 Kafka 集群，在任意时刻都有且只有一个控制器。官网上有个名为 activeController 的 JMX 指标，能够帮助咱们实时监控控制器的存活状态。这个JMX 指标很是关键，你在实际运维操做过程当中，必定要实时查看这个指标的值。下面，咱们就来详细说说控制器的原理和内部运行机制。

在开始以前，我先简单介绍一下 Apache ZooKeeper 框架。要知道，控制器是重度依赖ZooKeeper 的，所以，咱们有必要花一些时间学习下 ZooKeeper 是作什么的。

Apache ZooKeeper 是一个提供高可靠性的分布式协调服务框架。它使用的数据模型相似于文件系统的树形结构，根目录也是以“/”开始。该结构上的每一个节点被称为 znode，用来保存一些元数据协调信息。

若是以 znode 持久性来划分，znode 可分为持久性 znode 和临时 znode。持久性znode 不会由于 ZooKeeper 集群重启而消失，而临时 znode 则与建立该 znode 的ZooKeeper 会话绑定，一旦会话结束，该节点会被自动删除。

ZooKeeper 赋予客户端监控 znode 变动的能力，即所谓的 Watch 通知功能。一旦 znode节点被建立、删除，子节点数量发生变化，抑或是 znode 所存的数据自己变动，ZooKeeper 会经过节点变动监听器 (ChangeHandler) 的方式显式通知客户端。

依托于这些功能，ZooKeeper 常被用来实现集群成员管理、分布式锁、领导者选举等功能。Kafka 控制器大量使用 Watch 功能实现对集群的协调管理。咱们一块儿来看一张图片，它展现的是 Kafka 在 ZooKeeper 中建立的 znode 分布。你不用了解每一个 znode 的做用，但你能够大体体会下 Kafka 对 ZooKeeper 的依赖。

掌握了 ZooKeeper 的这些基本知识，如今咱们就能够开启对 Kafka 控制器的讨论了。

控制器是如何被选出来的？

你必定很想知道，控制器是如何被选出来的呢？咱们刚刚在前面说过，每台 Broker 都能充当控制器，那么，当集群启动后，Kafka 怎么确认控制器位于哪台 Broker 呢？

实际上，Broker 在启动时，会尝试去 ZooKeeper 中建立 /controller 节点。Kafka 当前选举控制器的规则是：第一个成功建立 /controller 节点的 Broker 会被指定为控制器。

控制器是作什么的？

咱们常常说，控制器是起协调做用的组件，那么，这里的协调做用究竟是指什么呢？我想了一下，控制器的职责大体能够分为 5 种，咱们一块儿来看看。

1. 主题管理（建立、删除、增长分区）

这里的主题管理，就是指控制器帮助咱们完成对 Kafka 主题的建立、删除以及分区增长的操做。换句话说，当咱们执行kafka-topics 脚本时，大部分的后台工做都是控制器来完成的。关于 kafka-topics 脚本，我会在专栏后面的内容中，详细介绍它的使用方法。

2. 分区重分配

分区重分配主要是指，kafka-reassign-partitions 脚本（关于这个脚本，后面我也会介绍）提供的对已有主题分区进行细粒度的分配功能。这部分功能也是控制器实现的。

3. Preferred 领导者选举

Preferred 领导者选举主要是 Kafka 为了不部分 Broker 负载太重而提供的一种换Leader 的方案。在专栏后面说到工具的时候，咱们再详谈 Preferred 领导者选举，这里你只须要了解这也是控制器的职责范围就能够了。

4.集群成员管理（新增 Broker、Broker 主动关闭、Broker 宕机）

这是控制器提供的第 4 类功能，包括自动检测新增 Broker、Broker 主动关闭及被动宕机。这种自动检测是依赖于前面提到的 Watch 功能和 ZooKeeper 临时节点组合实现的。

好比，控制器组件会利用Watch 机制检查 ZooKeeper 的 /brokers/ids 节点下的子节点数量变动。目前，当有新 Broker 启动后，它会在 /brokers 下建立专属的 znode 节点。一旦建立完毕，ZooKeeper 会经过 Watch 机制将消息通知推送给控制器，这样，控制器就能自动地感知到这个变化，进而开启后续的新增 Broker 做业。

侦测 Broker 存活性则是依赖于刚刚提到的另外一个机制：临时节点。每一个 Broker 启动后，会在 /brokers/ids 下建立一个临时 znode。当 Broker 宕机或主动关闭后，该 Broker 与ZooKeeper 的会话结束，这个 znode 会被自动删除。同理，ZooKeeper 的 Watch 机制将这一变动推送给控制器，这样控制器就能知道有 Broker 关闭或宕机了，从而进行“善后”。

5. 数据服务

控制器的最后一大类工做，就是向其余 Broker 提供数据服务。控制器上保存了最全的集群元数据信息，其余全部 Broker 会按期接收控制器发来的元数据更新请求，从而更新其内存中的缓存数据。

控制器保存了什么数据？

接下来，咱们就详细看看，控制器中到底保存了哪些数据。我用一张图来讲明一下。

怎么样，图中展现的数据量是否是不少？几乎把咱们能想到的全部 Kafka 集群的数据都囊括进来了。这里面比较重要的数据有：

• 全部主题信息。包括具体的分区信息，好比领导者副本是谁，ISR 集合中有哪些副本等。

• 全部 Broker 信息。包括当前都有哪些运行中的 Broker，哪些正在关闭中的 Broker 等。

• 全部涉及运维任务的分区。包括当前正在进行 Preferred 领导者选举以及分区重分配的分区列表。

值得注意的是，这些数据其实在 ZooKeeper 中也保存了一份。每当控制器初始化时，它都会从 ZooKeeper 上读取对应的元数据并填充到本身的缓存中。有了这些数据，控制器就能对外提供数据服务了。这里的对外主要是指对其余 Broker 而言，控制器经过向这些Broker 发送请求的方式将这些数据同步到其余 Broker 上。

控制器故障转移（Failover）

咱们在前面强调过，在 Kafka 集群运行过程当中，只能有一台 Broker 充当控制器的角色，那么这就存在单点失效（Single Point of Failure）的风险，Kafka 是如何应对单点失效的呢？答案就是，为控制器提供故障转移功能，也就是说所谓的 Failover。

故障转移指的是，当运行中的控制器忽然宕机或意外终止时，Kafka 可以快速地感知到，并当即启用备用控制器来代替以前失败的控制器。这个过程就被称为 Failover，该过程是自动完成的，无需你手动干预。

接下来，咱们一块儿来看一张图，它简单地展现了控制器故障转移的过程。

最开始时，Broker 0 是控制器。当 Broker 0 宕机后，ZooKeeper 经过 Watch 机制感知到并删除了 /controller 临时节点。以后，全部存活的 Broker 开始竞选新的控制器身份。Broker 3 最终赢得了选举，成功地在 ZooKeeper 上重建了 /controller 节点。以后，Broker 3 会从 ZooKeeper 中读取集群元数据信息，并初始化到本身的缓存中。至此，控制器的 Failover 完成，能够行使正常的工做职责了。

控制器内部设计原理

在 Kafka 0.11 版本以前，控制器的设计是至关繁琐的，代码更是有些混乱，这就致使社区中不少控制器方面的 Bug 都没法修复。控制器是多线程的设计，会在内部建立不少个线程。好比，控制器须要为每一个 Broker 都建立一个对应的 Socket 链接，而后再建立一个专属的线程，用于向这些 Broker 发送特定请求。若是集群中的 Broker 数量不少，那么控制器端须要建立的线程就会不少。另外，控制器链接 ZooKeeper 的会话，也会建立单独的线程来处理 Watch 机制的通知回调。除了以上这些线程，控制器还会为主题删除建立额外的I/O 线程。

比起多线程的设计，更糟糕的是，这些线程还会访问共享的控制器缓存数据。咱们都知道，多线程访问共享可变数据是维持线程安全最大的难题。为了保护数据安全性，控制器不得不在代码中大量使用ReentrantLock 同步机制，这就进一步拖慢了整个控制器的处理速度。

鉴于这些缘由，社区于 0.11 版本重构了控制器的底层设计，最大的改进就是，把多线程的方案改为了单线程加事件队列的方案。我直接使用社区的一张图来讲明。

从这张图中，咱们能够看到，社区引入了一个事件处理线程，统一处理各类控制器事件，而后控制器将原来执行的操做所有建模成一个个独立的事件，发送到专属的事件队列中，供此线程消费。这就是所谓的单线程 + 队列的实现方式。

值得注意的是，这里的单线程不表明以前提到的全部线程都被“干掉”了，控制器只是把缓存状态变动方面的工做委托给了这个线程而已。

这个方案的最大好处在于，控制器缓存中保存的状态只被一个线程处理，所以再也不须要重量级的线程同步机制来维护线程安全，Kafka 不用再担忧多线程并发访问的问题，很是利于社区定位和诊断控制器的各类问题。事实上，自 0.11 版本重构控制器代码后，社区关于控制器方面的 Bug 明显少多了，这也说明了这种方案是有效的。

针对控制器的第二个改进就是，将以前同步操做 ZooKeeper 所有改成异步操做。ZooKeeper 自己的 API 提供了同步写和异步写两种方式。以前控制器操做 ZooKeeper 使用的是同步的 API，性能不好，集中表现为，当有大量主题分区发生变动时，ZooKeeper容易成为系统的瓶颈。新版本 Kafka 修改了这部分设计，彻底摒弃了以前的同步 API 调用，转而采用异步 API 写入 ZooKeeper，性能有了很大的提高。根据社区的测试，改为异步以后，ZooKeeper 写入提高了 10 倍！

除了以上这些，社区最近又发布了一个重大的改进！以前 Broker 对接收的全部请求都是一视同仁的，不会区别对待。这种设计对于控制器发送的请求很是不公平，由于这类请求应该有更高的优先级。

举个简单的例子，假设咱们删除了某个主题，那么控制器就会给该主题全部副本所在的Broker 发送一个名为StopReplica的请求。若是此时 Broker 上存有大量积压的 Produce请求，那么这个 StopReplica 请求只能排队等。若是这些 Produce 请求就是要向该主题发送消息的话，这就显得很讽刺了：主题都要被删除了，处理这些 Produce 请求还有意义吗？此时最合理的处理顺序应该是，赋予 StopReplica 请求更高的优先级，使它可以获得抢占式的处理。

这在 2.2 版本以前是作不到的。不过自 2.2 开始，Kafka 正式支持这种不一样优先级请求的处理。简单来讲，Kafka 将控制器发送的请求与普通数据类请求分开，实现了控制器请求单独处理的逻辑。鉴于这个改进仍是很新的功能，具体的效果咱们就拭目以待吧。

总结

好了，有关 Kafka 控制器的内容，我已经讲完了。最后，我再跟你分享一个小窍门。当你以为控制器组件出现问题时，好比主题没法删除了，或者重分区 hang 住了，你不用重启Kafka Broker 或控制器。有一个简单快速的方式是，去 ZooKeeper 中手动删除/controller 节点。具体命令是 rmr /controller。这样作的好处是，既能够引起控制器的重选举，又能够避免重启 Broker 致使的消息处理中断。