[大数据]kafka核心组件控制器

控制器

在启动kafka集群的时候，每个代理都会实例化并且启动一个kafkacontroller，把代理的brokerid注册到zk的相应节点上。集群中的各个代理会根据选举机制选出leader控制器。当leader控制器发生宕机的时候其他代理会选出新的leader控制器。
控制器主要负责主题的创建和删除，分区和副本的管理，代理故障转移等处理。

1. 1、控制器初始化
   （1）创建controllercontext对象用于缓存控制器处理操作所需要得数据结构。数据结构包括：记录控制器选举次数得epoch以及对应的zkversion初始值都为0，当前正常运行的代理列表，主题列表，各主题对应的分区和副本的AR列表等。声明控制器和其他代理通信的controllerchannelmanage，实例化代理选举控制器操作的reetrantlock。
   （2）实例化用于维护和管理分区状态的状态机patitionstatemachine。分区定义了4种状态
   
   分区状态机会注册两个监听器TopicChangeListener和DeleteTopicListener。
   TopicChangeListener用于监听zk的目录下 /borkers/topics路径子节点的变化。当创建一个主题的时候会在该路径下创建一个和主题名称相同的子节点。当该路径下的子节点发生变化的时候就会触发该监听器。该监听器的方法handleChildChange()方法会更新controllerContext中的主题列表信息以及各主题对应的分区的AR信息。
   DeleteTopicListener用于监听 zk的目录 /admin/delete_topics子节点的变更。当删除一个主题时，会在该节点下创建一个与删除主题名称相同的子节点。当该路径下的子节点发生变化的时候就会触发这个监听器，该监听器的han’d’leChildChange()方法会将待删除的主题从路径 /borkers/topics路径下删除，将该主题加入到待删除的主题队列中交给TopicDeletionManager执行删除。
   （3）实例化一个对分区副本状态管理的状态机ReplicaStateMachine。副本状态有7种
   
   在副本状态机的内部定义了一个BrokerChangeListener监听器，该监听器会监听zk的目录 /broker/ids路径下的各代理的brokerid节点的变化。如果代理与zk会话超时，或者是有新代理加入 /broker/ids节点下就会发生变化就会触发该监听器，该监听器调用controllercontext.controllerChannleManager对节点变化进行处理。
   （4）创建选举对象ZooKeeperLeaderElectoer。实例化该对象的时候需要传入2个回调函数。完成控制器初始化操作的onControllerFailover()方法，当新的控制器当选时，让之前的控制器注销控制权限onControllerResignation()方法。
   （5）创建一个独立的定时任务KafkaScheduler，用于控制器平衡操作。只是在代理是leader控制器的时候有效，如果代理不是leader的时候也就是调用onControllerResignation()方法时就会关闭该定时任务。
   （6）声明一个对主题管理的TopicDelitionManager
   （7）创建一个用于在分区状态发生变化时为分区选举出leader副本的分区选举器patitionLeaderSelector。kafka提供5种分区选举器
   各选举器的功能及选举策略如下。
   E俨Selecto俨
   ? OffiinePartitionLeaderSelector： 分区状态机启动、新创建一个分区或是将一个分区状态 由 NewPartition 状态、 OffiinePartition 状态转换到 OnlinePartition 状态时会调用该选举 器， 为分区选出 Leader， 得到分区的 LeaderAndlsr。该选举器选举策略是首先判断是否 有存活（本书用存活来代表心跳检测正常的代理所处的状态）的 ISR，若 ISR 中至少有 一个存活的代理，则从 ISR 列表中选第一个存活的代理作为 Leader，存活的 ISR 作为 新的 ISR； 否则， 若配置项 unclean.leader.election.enable 为 位ue，该配置项默认为 true,
   3.2 控制器 45
   即表示允许从不在 JSR 列表中的副本选举 Leader，同时 AR 中若有存活的副本，则从 AR 列表中选第一个代理作为 Leader，存活的 AR 作为新的 JSR。当没有可选作 Leader 的代理时，会抛出 NoReplicaOnlineException 的异常。 Leader 和 JSR 选出后构造 LeaderAndlsr 对象，将当前的 leader_epoch 加 l 赋值给新的 leader_epoch，将当前的 zk:Version 加 l 作为新的 zkVersion。若 Leader 选举成功，后续会将 LeaderAndisr 对象和 controller_ epoch 值构造 PartitionStatelnfo 对象。登录 ZooKeeper 客户端在 ZooKe巳per 的 /brokers此topics/$topicName/partitions/$ {partitionld }/state 可以查看到某个分区的元数据信 息、。例如，查看有 5 个 Broker，即 brokerld={0,1,2,3,4｝，主题名 topicName 为“topic-analyse”， partitionld 为 0 的分区信息如下：
   ｛ ” controller_epoch “:7 ，” leader”: 3 ，”version":l ，” leader_epoch ”： 0 ，”工sr ”：［ 1, 3]} 以上分区信息表示该分区的 Leader 副本为 brokerld 为 3 的节点，同时该分区至少有 2 个副 本，因为 ISR 当前有 2 个节点，集群的控制器总共发生了 7 次变更，而分区的 Leader 从最 初当选还未发生过变化，可以从 leader_epoch 值来判断分区或是 ISR 是否发生过变化。 ? ReassignedPartitionLeaderSelector： 当分区进行重分配时会调用该选举器。该选举器的 选举策略是从 AR 列表中找出存活副本列表，若有存活的副本则取存活副本列表的第一 个副本作为 Leader，将当前 ISR 作为新的 JSR， 将 AR 作为接受 LeaderAndlsr 请求的副 本集合。若没有候选的副本，则抛出 NoReplicaOnlineException 异常。 PreferredReplicaPartitionLeaderSelector：该选举器直接将优先副本设置为分区的 Leader。 该选举器首先根据当前 Leader 是不是由优先副本担任来决定是否需要选举。若当前 Leader 由优先副本担任则无需设置，仅抛出 LeaderElectionNotNeededException 异常进行 提示：若优先副本不是 Leader 但在该分区的 ISR 列表中，则将优先副本选为 Leader，将 AR 作为接受 LeaderAndisr 请求的副本集合：否则抛出 StateChangeFailedException 异常。 ? ControlledShutdownLeaderSelector： 该选举器将从 ISR 中剔除己关闭的节点，将剔除己 关闭节点后的 ISR 作为新的 ISR，同时从新的 ISR 中选取第一个作为 Leader 副本。将 AR 中剔除己关闭节后的副本节点作为接受 LeaderAndlsr 请求的副本集合。 ? NoOpLeaderSelector：该选举器只返回当前分区的 Leader 和 ISR。
   （8）实例化 ControllerBrokerRequestBatch。在前面实例化了分区状态机和副本状态机，这 两个状态机在相应状态发生变化时相应监听器都会调用各自的 handleStateChange（）方法进行处 理，而 ControllerBrokerRequestBatch 封装了 leaderAndisrRequestMap、 stopReplicaRequestMap 和 updateMetadataRequestMap 这 3 个集合，用来记录和缓存 handleStateChange（）方法中产生的 request，控制器将这些 request 交由 ControllerBrokerRequestBatch.sendRequestsToBrokers（）方法 批量发送出去，交由 KafkaApis 调用相应的 handle 方法进行处理
   （9）实例化 3 个监昕器，即用于监昕分区重分配的 PartitionsReassignedListener，用于监昕 当分区状态变化时触发 PreferredReplicaPartitionLeaderSelector 选举器将优先副本选举为 Leader 的 PreferredReplicaElectionListener、用于监听当 ISR 发生变化时将 ISR 变化通知给 ZooKeeper
   进行更新操作，同时向所有的代理节点发送元数据修改请求的 IsrChangeNotificationListener。
   。当一个代理启动时就会创建一个 KafkaController 实例 并启动。在启动 KafkaCon位oiler 时，先注册一个用于监听代理与 ZooKeeper 会话超时的监听器 SessionExpirationListener，然后启动控制器选举，让当前代理试图去竞选为控制器。