Kafka技术详解[5]: 集群启动
目录
集群启动
相关概念
代理：Broker
控制器：Controller
启动ZooKeeper
启动Kafka
初始化ZooKeeper
初始化服务
启动任务调度器
创建数据管理器
创建远程数据管理器
创建副本管理器
创建ZK元数据缓存
创建Broker通信对象
创建网络通信对象
注册Broker节点
启动控制器
初始化通道管理器
初始化事件管理器
初始化状态管理器
启动控制器
集群启动
在搭建完Kafka集群之后，我们需要了解一些基本的概念，并掌握如何正确地启动集群。以下是一些关键的概念介绍以及启动步骤。
相关概念
代理：Broker
在使用Kafka之前，我们需要启动Kafka服务进程，这个服务进程通常被称为Kafka Broker或Kafka Server。由于Kafka是一个分布式消息系统，在生产环境中，通常需要多个服务进程组成集群来提供消息服务。因此，每一个服务节点都称为一个Broker，并且为了区分不同的服务节点，每一个Broker都需要有一个唯一的全局ID，即broker.id，这个ID可以在Kafka软件的配置文件server.properties中进行配置。
示例配置如下：
# The id of the broker. This must be set to a unique integer for each broker
# 集群ID
broker.id=0在我们的Kafka集群中，每个节点都有自己的ID，是一个唯一的整数。
| 主机 | kafka-broker1 | kafka-broker2 | kafka-broker3 |
|---|---|---|---|
| broker.id | 1 | 2 | 3 |
控制器：Controller
Kafka作为一个分布式消息传输系统，存在多个Broker服务节点，并且采用了常见的主从（Master-Slave）架构。这意味着需要从多个Broker中选出一个用于管理整个Kafka集群的Master节点，这个节点称为Controller。Controller是Kafka的核心组件之一，其主要职责是在Apache Zookeeper的帮助下管理和协调整个Kafka集群。
如果在运行过程中，Controller节点出现故障，Kafka会依赖于ZooKeeper软件来选举出新的Controller，以确保集群的高可用性。
Kafka集群中Controller的基本功能包括但不限于：
- Broker管理：监听
/brokers/ids节点相关的所有变化，包括Broker数量的增减以及Broker对应数据的变化。 - Topic管理：监听
/brokers/topics节点相关的所有变化，包括Topic的新增、修改和删除操作。 - Partition管理：监听
/admin/reassign_partitions、/isr_change_notification、/preferred_replica_election等节点相关的所有变化。 - 数据服务：启动分区状态机和副本状态机。
启动ZooKeeper
Kafka集群含有多个服务节点，而在经典的主从架构中，需要从多个服务节点中选出一个作为集群管理的Master节点，即Controller。如果Controller节点出现故障，则需要从其余的Slave节点中选举出一个新的Controller节点来接管管理功能。
Kafka依赖于ZooKeeper软件来实现Broker节点的选举功能。具体来说，ZooKeeper提供了以下功能支持：
- 创建节点：创建一个节点时，可以选择持久化创建或临时创建。持久化创建的节点会一直存在，而临时创建的节点则依赖于客户端的连接状态，一旦客户端断开连接，节点也会被自动删除。
- 节点唯一性：ZooKeeper中的节点不允许重复创建，因此多个客户端尝试创建相同的节点时，只有第一个成功创建的客户端可以创建节点。
- 监听机制：客户端可以在ZooKeeper节点上设置监听器，以便在节点状态发生变化时收到通知并作出响应。
Kafka利用ZooKeeper的这些特性来实现Controller节点的选举：
- 当首次启动Kafka集群时，多个Broker节点会同时启动并尝试连接ZooKeeper，每个Broker节点都会尝试创建一个临时节点
/controller。 - 由于ZooKeeper中一个节点不允许重复创建，因此最终只有一个Broker节点能够成功创建
/controller节点，这个Broker节点将成为Kafka集群的Controller，负责管理整个集群。 - 其他未成为Controller的Broker节点会在
/controller节点上设置监听器，以监听其状态变化。 - 如果Controller节点发生故障导致连接中断，
/controller节点会被自动删除，此时其他设置了监听器的Broker节点会检测到这一变化，并尝试重新创建/controller节点，成功者将成为新的Controller。
启动Kafka
初始化ZooKeeper
在Kafka Broker启动期间，首先会创建一个ZooKeeper客户端（KafkaZkClient）以与ZooKeeper进行交互。创建完客户端对象后，它会向ZooKeeper发送创建节点的请求，这些节点都是持久化的。以下是一些重要的节点及其作用：
/admin/delete_topics：持久化节点，用于配置待删除的主题。因删除过程可能涉及Broker下线或失败情况，因此需要在Broker重新上线后根据此节点继续删除操作。一旦主题的所有分区数据被删除，则清理该节点。/brokers/ids：持久化节点，存储服务节点ID标识。每当有新的Broker启动，就在该节点下增加子节点。Broker ID必须唯一。/brokers/topics：持久化节点，包含服务节点中的主题详情，如分区和副本。/brokers/seqid：持久化节点，用于自动生成Broker ID。/config/changes：持久化节点，当Kafka元数据发生变化时，在该节点下创建子节点。/config/clients：持久化节点，存储客户端配置，默认为空。/config/brokers：持久化节点，存储服务节点相关配置，默认为空。/config/ips：持久化节点，存储IP配置，默认为空。/config/topics：持久化节点，存储主题配置，默认为空。/config/users：持久化节点，存储用户配置，默认为空。/consumers：持久化节点，记录消费者相关信息。/isr_change_notification：持久化节点，用于通知Controller及时更新ISR列表。/latest_producer_id_block：持久化节点，存储PID块，确保生产者请求得到响应。/log_dir_event_notification：持久化节点，当Broker中的数据路径出现问题时，向ZooKeeper添加通知序号。/cluster/id：持久化节点，存储Kafka集群的唯一ID信息及其版本号。
初始化服务
启动任务调度器
每个Broker启动时都会创建一个内部调度器（KafkaScheduler）并启动，使用Java中的定时任务线程池ScheduledThreadPoolExecutor来完成节点内部的工作任务。
创建数据管理器
每个Broker启动时创建数据管理器（LogManager），用于接收消息后的数据创建、查询、清理等处理。
创建远程数据管理器
每个Broker启动时创建远程数据管理器（RemoteLogManager），用于与其他Broker节点进行数据状态同步。
创建副本管理器
每个Broker启动时创建副本管理器（ReplicaManager），负责处理主题的副本。
创建ZK元数据缓存
每个Broker启动时创建ZK元数据缓存（ZkMetadataCache），用于缓存ZK中的Kafka元数据。
创建Broker通信对象
每个Broker启动时创建Broker间通信管理器（BrokerToControllerChannelManager），管理Broker与Controller间的通信。
创建网络通信对象
每个Broker启动时创建网络通信对象（SocketServer），用于与其他Broker进行通信，包含Java NIO通信的Channel和Selector对象。
注册Broker节点
Broker启动时，通过ZK客户端向ZK注册当前Broker节点ID，创建临时节点。如果Broker与ZK的连接断开，则临时节点会被删除。
启动控制器
控制器（KafkaController）是每个Broker启动时创建的核心对象，用于与ZK建立连接并申请成为整个Kafka集群的管理者。如果申请成功，它会初始化管理器并建立与其他Broker的数据通道。
初始化通道管理器
创建通道管理器（ControllerChannelManager），维护Controller与集群所有Broker节点间的网络连接，并向Broker发送控制请求及接收响应。
初始化事件管理器
创建事件管理器（ControllerEventManager），用于管理Controller与集群所有Broker节点间的网络连接，并向Broker发送控制请求及接收响应。
初始化状态管理器
创建状态管理器（ControllerChangeHandler），可监听/controller节点的操作，并在节点创建、删除或数据变化时执行相应处理。
启动控制器
控制器对象启动后，向事件管理器发送Startup事件。事件处理线程接收到事件后，通过ZK客户端向ZK申请/controller节点。申请成功后，执行一系列使当前节点成为Controller的操作，包括注册ZooKeeper监听器、删除日志路径变更及ISR副本变更通知事件、启动Controller通道管理器，以及启动副本状态机和分区状态机。