1、导读
KAFKA 是基于 Scala 语言开发的一个多分区、多副本且基于 ZooKeeper 协调的分布式消息系统,它以高吞吐、可持久化、可水平扩展、支持流数据处理等多种特性而被广泛使用,越来越多的开源分布式处理系统如 Cloudera、Storm、Spark、Flink 等都支持与 KAFKA 集成。本文将基于 KAFKA v1.1.0 版本源码,探讨 KafkaController 的启动流程、选举流程、脑裂问题和事件队列模型。笔者水平有限,若有不当之处,敬请指正。
2、Controller 是什么
很多时候业务为了保证 7x24 小时提供服务,假如一台机器宕机,仍然能有其它机器顶替它继续工作,一般采用 Leader/Follower 模式,即常说的主从模式。正常情况下主机提供服务,备机只负责监听主机状态,当主机故障时,业务可以自动切换到备机继续提供服务,而在切换过程中从备机中选出主机的过程就是 Leader 选举。
KAFKA 集群的多个服务代理节点(节点,Broker)中,有一个 Broker 会被选举为控制器(Controller)。Controller 负责管理整个集群中所有分区和副本的状态变化,并时刻检测集群状态的变化,例如 Broker 故障、分区的 Leader 副本选举和分区重分配。每个 KAFKA 集群只有一个 Controller,以维护集群的单一一致视图。虽然 Controller 成为单点,但 KAFKA 有处理单点故障的机制。
3、Controller 的启动流程
KAFKA Server 启动时会做一系列的初始化,例如:初始化 ZooKeeper、启动 KafkaController、启动 GroupCoordinator、启动 ReplicaManager 等等。
3.1、初始化 ZooKeeper
初始化 ZooKeeper 主要是建立连接,注册监听器,其中值得关注的是在初始化原生ZooKeeper对象的同时会注册一个全局的事件监听器(Watcher),负责监听ZK节点数据变更、ZK的会话状态等等。
// sessionTimeoutMs: server.properties 配置文件中的 zookeeper.session.timeout.ms
@volatile private var zooKeeper = new ZooKeeper(connectString, sessionTimeoutMs, ZooKeeperClientWatcher)
ZooKeeperClientWatcher 负责监听ZK节点数据变更、会话状态等等。
private[zookeeper] object ZooKeeperClientWatcher extends Watcher {
override def process(event: WatchedEvent): Unit = {
debug(s"Received event: $event")
Option(event.getPath) match {
case None =>
val state = event.getState
stateToMeterMap.get(state).foreach(_.mark())
inLock(isConnectedOrExpiredLock) {
isConnectedOrExpiredCondition.signalAll()
}
if (state == KeeperState.AuthFailed) {
error("Auth failed.")
stateChangeHandlers.values.foreach(_.onAuthFailure())
} else if (state == KeeperState.Expired) {
// 监听Broker与ZK的会话过期事件
scheduleSessionExpiryHandler()
}
case Some(path) =>
// 监听ZK节点状态,将事件分发至相应的Handler
(event.getType: @unchecked) match {
case EventType.NodeChildrenChanged => zNodeChildChangeHandlers.get(path).foreach(_.handleChildChange())
case EventType.NodeCreated => zNodeChangeHandlers.get(path).foreach(_.handleCreation())
case EventType.NodeDeleted => zNodeChangeHandlers.get(path).foreach(_.handleDeletion())
case EventType.NodeDataChanged => zNodeChangeHandlers.get(path).foreach(_.handleDataChange())
}
}
}
}
其中 zNodeChangeHandlers、zNodeChildChangeHandlers 均是 ConcurrentHashMap 结构,在后续的 KAFKA 的启动过程中会将相应的 Handler 添加进来,ZK节点监听器监听到节点发生变化后,KafkaController 负责从 ConcurrentHashMap 结构中根据 ZK 节点路径取出对应的处理器(Handler)处理事件。
private val zNodeChangeHandlers = new ConcurrentHashMap[String, ZNodeChangeHandler]().asScala
private val zNodeChildChangeHandlers = new ConcurrentHashMap[String, ZNodeChildChangeHandler]().asScala
3.2、启动内存队列异步事件管理器
所有 Broker 会定期检查 ZooKeeper 以表明自己是健康的。如果 Broker 的响应时间超过 zookeeper.session.timeout.ms(默认18,000 毫秒),则意味着 Broker 与 ZooKeeper 的会话过期,随后将触发 Controller 选举。
eventManager 的对象类型是 ControllerEventManager,用于统一管理监听器触发事件,当ZK监听到节点变化后,其运用单线程方式从内存队列中取出对应的 Handler 处理事件。「Controller 的事件队列模型」章节会展开讲述。
def startup() = {
zkClient.registerStateChangeHandler(new StateChangeHandler {
override val name: String = StateChangeHandlers.ControllerHandler
override def beforeInitializingSession(): Unit = {
// 初始化 KafkaController 和 ZK 的会话信息
// 1、重置内存中的ActiveControllerId
// 2、重置内存中的依赖对象信息,例如监听器触发的事件、上下文信息
// 3、重置Kafka Broker状态
val expireEvent = new Expire
eventManager.clearAndPut(expireEvent)
expireEvent.waitUntilProcessed()
}
override def afterInitializingSession(): Unit = {
// 事件管理器添加「注册Broker和选举Controller」事件
// 1、ZK注册Broker信息
// 2、选举Controller
eventManager.put(RegisterBrokerAndReelect)
}
})
// 事件管理器添加KafkaController的启动事件
eventManager.put(Startup)
// 单线程启动事件管理器
eventManager.start()
}
3.3、选举 Controller
每个 Broker 启动过程中,都会尝试在 ZK 创建 /controller 临时节点,也有可能其它 Broker 同时尝试创建,只有创建成功的那个 Broker 才会成为 Controller,而创建失败的 Broker 选举失败。「Controller 的选举流程」章节会展开讲述。
case object Startup extends ControllerEvent {
...
override def process(): Unit = {
// 1、/controller 节点添加ControllerChangeHandler,监听Controller的变更事件
zkClient.registerZNodeChangeHandlerAndCheckExistence(controllerChangeHandler)
// 2、选举Controller
elect()
}
}
3.3、初始化上下文信息
Controller 在选举成功后会读取 ZK 中各节点的数据来完成一系列的初始化工作,包括初始化上下文信息(从ZK读取ControllerEpoch、Topic、分区、副本相关的各种元数据信息)、启动并管理分区和副本的状态机、定义监听器触发事件、定义周期性任务触发事件(如分区自平衡)。不管是监听器触发的事件,还是周期性任务触发事件,都会读取或更新 Controller 中的上下文信息。
private def onControllerFailover() {
// 1、读取 /controller_epoch 的值到 ControllerContext(上下文信息)
// 2、递增 ControllerEpoch,并更新到 /controller_epoch
readControllerEpochFromZooKeeper()
incrementControllerEpoch()
// 3、注册 ZK 监听器触发事件(逻辑上基于事件队列模型实现,并非真正在ZK上增加Watcher)
// - Broker 相关变化的事件处理器
// - Topic 相关变化的事件处理器
// - ISR 集合变更的事件处理器
// - 优先副本选举的事件处理器
// - 分区重分配的事件处理器
val childChangeHandlers = Seq(brokerChangeHandler, topicChangeHandler, topicDeletionHandler, logDirEventNotificationHandler,
isrChangeNotificationHandler)
childChangeHandlers.foreach(zkClient.registerZNodeChildChangeHandler)
val nodeChangeHandlers = Seq(preferredReplicaElectionHandler, partitionReassignmentHandler)
nodeChangeHandlers.foreach(zkClient.registerZNodeChangeHandlerAndCheckExistence)
...
// 4、初始化上下文信息,ControllerContext用于维护Topic、分区和副本相关的各种元数据,
//
initializeControllerContext()
...
}
private def initializeControllerContext() {
// 读取 /brokers/ids,初始化可用 Broker 集合
controllerContext.liveBrokers = zkClient.getAllBrokersInCluster.toSet
// 读取 /broker/topics,初始化集群中所有 Topic 集合
controllerContext.allTopics = zkClient.getAllTopicsInCluster.toSet
// 所有 Topic 对应ZK中 /brokers/topics/<topic> 节点添加PartitionModificationsHandler,用于监听Topic中的分区分配变化
registerPartitionModificationsHandlers(controllerContext.allTopics.toSeq)
// 读取 /broker/topics/<topic>/partitions,初始化每个Topic每个分区的AR集合
controllerContext.partitionReplicaAssignment = mutable.Map.empty ++ zkClient.getReplicaAssignmentForTopics(controllerContext.allTopics.toSet)
controllerContext.partitionLeadershipInfo = new mutable.HashMap[TopicPartition, LeaderIsrAndControllerEpoch]
controllerContext.shuttingDownBrokerIds = mutable.Set.empty[Int]
// 所有可用 Broker 对应ZK中 /brokers/ids/<id> 节点添加BrokerModificationsHandler,用于监听Broker增减的变化
registerBrokerModificationsHandler(controllerContext.liveBrokers.map(_.id))
// 初始化每个分区的 Leader、ISR 集合等信息
updateLeaderAndIsrCache()
// 启动 ControllerChannelManager
startChannelManager()
// 读取 /admin/reassign_partitions,初始化需要进行副本重分配的分区
initializePartitionReassignment()
...
}
KafkaController 相关的 ZooKeeper 路径如下:
-
/controller :存放 Controller 的信息。记录 Controller 所在的 brokerid,节点类型是临时节点,与ZK会话失效时自动删除。ControllerChangeHandler 监听 Controller 的变更事件;
-
/brokers/ids :存放集群中所有可用的 Broker。BrokerChangeHandler 负责处理 Broker 的增减事件;
-
/brokers/topics :存放集群中所有的 Topic。TopicChangeHandler 负责处理 Topic 的增减事件;
-
/admin/delete_topics :存放删除的 Topic。TopicDeletionHandler 负责处理 Topic 的删除事件;
-
/brokers/topics/<topic>/partitions :存放 Topic 的分区。PartitionModificationsHandler 负责处理分区分配事件;
-
/brokers/topics/<topic>/partitions/<partition>/state :存放的是分区的 Leader 副本和 ISR 集合信息。PartitionReassignmentIsrChangeHandler 负责在副本分配到分区时,需要等待新副本追上Leader 副本后才能执行后续操作;
-
/admin/reassign_partitions :存放重新分配AR的路径,通过命令修改AR时会写入此路径。PartitionReassignmentHandler 负责执行分区重新分配;
-
/admin/preferred_replica_election :存放需要选举副本 Leader 的分区信息。PreferredReplicaElectionHandler 负责对指定分区进行有限副本选举。
4、Controller 的选举流程
KafkaController 触发选举的时机有三个:
-
KafkaController 启动时;
-
KafkaController 与 ZooKeeper 的会话过期;
-
ZooKeeper 中 /controller 临时节点不存在时;
每个 Broker 启动过程中,都会尝试在 ZK 创建 /controller 临时节点,同时也有可能其它 Broker 尝试创建,但只有创建成功的那个 Broker 才会成为 Controller,而创建失败的 Broker 选举失败。图 4-1 展示了 Broker0 成功创建临时节点并写入 Controller 信息;图 4-2 根据展示了当 Broker0 竞选成为 Controller,而 Broker1 和 Broker 2 竞选失败后作为 Follower 后不工作;
图 4-1
图 4-2
每个 Broker 都会在内存中保存当前 Controller 的 brokerid 的值,标识为 activeControllerId。
private def elect(): Unit = {
...
// 步骤1、获取ZK的/controller临时节点中brokerid的值,若节点不存在则取-1
activeControllerId = zkClient.getControllerId.getOrElse(-1)
if (activeControllerId != -1) {
debug(s"Broker $activeControllerId has been elected as the controller, so stopping the election process.")
return
}
try {
// 步骤2:尝试在ZK中创建/controller临时节点,若创建成功则竞选成为Controller
zkClient.checkedEphemeralCreate(ControllerZNode.path, ControllerZNode.encode(config.brokerId, timestamp))
info(s"${config.brokerId} successfully elected as the controller")
activeControllerId = config.brokerId
// 步骤3:更新/controller_epoch等信息
onControllerFailover()
} catch {
...
}
}
private def onControllerFailover() {
// 步骤4、读取 /controller_epoch 的值到 ControllerContext(上下文信息)
// 步骤5、递增 ControllerEpoch,并更新到 /controller_epoch
info("Reading controller epoch from ZooKeeper")
readControllerEpochFromZooKeeper()
info("Incrementing controller epoch in ZooKeeper")
incrementControllerEpoch()
info("Registering handlers")
...
}
4.1、旧版的脑裂问题
早期 KAFKA 版本(例如 0.8.x)的脑裂问题 ISSUE:KAFKA-Split-Brain
上文介绍过,KAFKA 是依赖 ZooKeeper 来做 Leader/Follower HA:每个节点都尝试注册一个象征 Leader 的临时节点,其他未注册成功的则成为 Follower,并且通过监听器机制监听着 Leader 所创建的临时节点,ZooKeeper 通过内部心跳机制来确定 Leader 的状态,一旦 Leader 出现故障,ZooKeeper 能很快获悉并且通知其他的 Follower,其他 Follower 尝试竞选,这样就完成了一个新的 Leader 切换。但值得注意的是,短暂的时间内可能由于 Leader 的 GC 时间过长或者 ZooKeeper 节点间网络抖动导致心跳超时(Leader 假死),ZooKeeper 与 Leader 的会话超时随后 ZooKeeper 通知剩余的 Follower 重新竞选,Follower 中就有一个成为了“新Leader”,但“旧Leader”并未故障,此时可能有一部分的客户端已收到 ZooKeeper 的通知并连接到“新Leader”,有一部分客户端仍然连接在“旧Leader”,如果同时两个客户端分别对新旧 Leader 的同一个数据进行更新,就会出现很严重问题。
ZooKeeper 会话过期后通知 Follower 重新竞选的代码入口:kafka.zookeeper.ZooKeeperClient#scheduleSessionExpiryHandler
图 4-3 展示了一种可能发生脑裂过程,Broker 0 是 Controller,但由于负载过高发生长时间 FGC,耗时超过 zookeeper.session.timeout.ms 导致与 ZooKeeper 的心跳超时,ZooKeeper 监听到会话过期事件并通知 Broker 1 和 Broker 2,Broker 1 竞选成功成为新的 Leader(Controller),Broker 2 竞选失败继续成为 Follower,随后 Broker 0 GC 结束仍然以 Leader(Controller) 身份运行,此时集群中就有多个 Leader(Controller);
图 4-3
为了解决脑裂问题,新版本的 KAFKA 中新增了“控制器纪元”(ZooKeeper 中是持久节点,路径是 /controller_epoch),Integer 类型,用于跟踪并记录 Controller 发生变更的次数,表示当前的 Controller 是第几代控制器。controller_epoch 的初始值为 1,当 Controller 发生变更,每选出一个新的 Controller 就将该字段自增。每个和 Controller 交互的请求都会携带 controller_epoch 字段:
-
若请求的 controller_epoch 值小于内存中的 epoch 值,则认为这个请求是发送给已过期的 Controller,那么这个请求会被认为是无效;
-
若请求的 controller_epoch 值大于内存中的 epoch 值,则说明已经有新的 Controller 当选;
由此可见,KAFKA 通过 controller_epoch 来保证 Controller 的唯一性;
5、Controller 的事件队列模型
从上文可知,Controller 在选举成功后会读取 ZK 中各节点的数据来完成一系列的初始化工作,不管是监听器触发的事件,还是周期性任务触发事件,都会读取或更新 Controller 中的上下文信息,这样就涉及到了多线程同步,如果单纯使用锁机制来控制,那么整体性能会大打折扣,实际上,在 0.10 版本之前,都是通过锁机制的方式来处理上下文的多线程同步,在 0.11+ 版本以后改用单线程基于事件队列的模型,将每个事件都做一层封装,然后按事件发生的先后顺序暂存到 LinkedBlockingQueue 中,最后使用 ControllerEventThread 线程按照 FIFO 的原则来处理各个事件,这样不需要锁机制就可以在多线程间维护线程安全。
class ControllerEventManager(controllerId: Int, rateAndTimeMetrics: Map[ControllerState, KafkaTimer],
eventProcessedListener: ControllerEvent => Unit) {
...
private val putLock = new ReentrantLock()
private val queue = new LinkedBlockingQueue[ControllerEvent]
private val thread = new ControllerEventThread(ControllerEventManager.ControllerEventThreadName)
def start(): Unit = thread.start()
def put(event: ControllerEvent): Unit = inLock(putLock) {
queue.put(event)
}
class ControllerEventThread(name: String) extends ShutdownableThread(name = name, isInterruptible = false) {
override def doWork(): Unit = {
queue.take() match {
...
case controllerEvent =>
try {
rateAndTimeMetrics(state).time {
controllerEvent.process()
}
} catch {
case e: Throwable => error(s"Error processing event $controllerEvent", e)
}
...
}
}
}
}
在 KAFKA 早期版本中,并没有采用 KafkaController 这样一个概念来统一管理分区和副本的状态,而是依赖ZK,每个 Broker 都会在 ZK 上为分区和副本注册大量的 Watcher,当分区或副本状态变化时,会唤醒很多不必要的 Watcher,这种严重依赖 ZK 的设计会有脑裂、羊群效应,以及造成 ZK 过载的隐患;在 0.11+ 版本以后,只有 KafkaController 在 ZK 上注册相应的 Watcher,其它的 Broker 极少需要再监听 ZK 中的数据变化,不过每个 Broker 还是会对 /controller 节点添加监听器,用于监听 Controller 是否存活;
图 5-1