当前位置: 首页 > news >正文

kafka集群Controller竞选与责任设计思路架构详解-kafka 商业环境实战

本套技术专栏是作者(秦凯新)平时工作的总结和升华,通过从真实商业环境抽取案例进行总结和分享,并给出商业应用的调优建议和集群环境容量规划等内容,请持续关注本套博客。期待加入IOT时代最具战斗力的团队。QQ邮箱地址:1120746959@qq.com,如有任何学术交流,可随时联系。

1 无所不能的Controller

  • 某一个broker被选举出来承担特殊的角色,就是控制器Controller。

  • Leader会向zookeeper上注册Watcher,其他broker几乎不用监听zookeeper的状态变化。

  • Controller集群就是用来管理和协调Kafka集群的,具体就是管理集群中所有分区的状态和分区对应副本的状态。

  • 每一个Kafka集群任意时刻都只能有一个controller,当集群启动的时候,所有的broker都会参与到controller的竞选,最终只能有一个broker胜出。

  • Controller维护的状态分为两类:1:管理每一台Broker上对应的分区副本。2:管理每一个Topic分区的状态。

  • KafkaController 核心代码,其中包含副本状态机和分区状态机

      class KafkaController(val config : KafkaConfig, zkClient: ZkClient, 
      val brokerState: BrokerState) extends Logging with KafkaMetricsGroup {
          this.logIdent = "[Controller " + config.brokerId + "]: "
          private var isRunning = true
          private val stateChangeLogger = KafkaController.stateChangeLogger
          val controllerContext = new ControllerContext(zkClient, config.zkSessionTimeoutMs)
    
          val partitionStateMachine = new PartitionStateMachine(this)
          val replicaStateMachine = new ReplicaStateMachine(this)
          
          private val controllerElector = new ZookeeperLeaderElector(controllerContext, ZkUtils.ControllerPath, onControllerFailover,
          onControllerResignation, config.brokerId)
          // have a separate scheduler for the controller to be able to start and stop independently of the
          // kafka server
          private val autoRebalanceScheduler = new KafkaScheduler(1)
          var deleteTopicManager: TopicDeletionManager = null
          val offlinePartitionSelector = new OfflinePartitionLeaderSelector(controllerContext, config)
          private val reassignedPartitionLeaderSelector = new ReassignedPartitionLeaderSelector(controllerContext)
          private val preferredReplicaPartitionLeaderSelector = new PreferredReplicaPartitionLeaderSelector(controllerContext)
          private val controlledShutdownPartitionLeaderSelector = new ControlledShutdownLeaderSelector(controllerContext)
          private val brokerRequestBatch = new ControllerBrokerRequestBatch(this)
          
          private val partitionReassignedListener = new PartitionsReassignedListener(this)
          private val preferredReplicaElectionListener = new PreferredReplicaElectionListener(this)
    复制代码
  • KafkaController中共定义了五种selector选举器

      1、ReassignedPartitionLeaderSelector
      从可用的ISR中选取第一个作为leader,把当前的ISR作为新的ISR,将重分配的副本集合作为接收LeaderAndIsr请求的副本集合。
      2、PreferredReplicaPartitionLeaderSelector
      如果从assignedReplicas取出的第一个副本就是分区leader的话,则抛出异常,否则将第一个副本设置为分区leader。
      3、ControlledShutdownLeaderSelector
      将ISR中处于关闭状态的副本从集合中去除掉,返回一个新新的ISR集合,然后选取第一个副本作为leader,然后令当前AR作为接收LeaderAndIsr请求的副本。
      4、NoOpLeaderSelector
      原则上不做任何事情,返回当前的leader和isr。
      5、OfflinePartitionLeaderSelector
      从活着的ISR中选择一个broker作为leader,如果ISR中没有活着的副本,则从assignedReplicas中选择一个副本作为leader,leader选举成功后注册到Zookeeper中,并更新所有的缓存。
    复制代码
  • kafka修改分区和副本数

      ../bin/kafka-topics.sh --zookeeper 127.0.0.1:2181 --describe  --topic test1
      
      Topic:test1       PartitionCount:3        ReplicationFactor:2     Configs:
      Topic: test1      Partition: 0    Leader: 2       Replicas: 2,4   Isr: 2,4
      Topic: test1      Partition: 1    Leader: 3       Replicas: 3,5   Isr: 3,5
      Topic: test1      Partition: 2    Leader: 4       Replicas: 4,1   Isr: 4,1
    复制代码
  • topic 分区扩容

      ./kafka-topics.sh --zookeeper 127.0.0.1:2181 -alter --partitions 4 --topic test1
    复制代码

2 ReplicaStateMachine (ZK持久化副本分配方案)

  • Replica有7种状态:

      1 NewReplica: 在partition reassignment期间KafkaController创建New replica
      
      2 OnlineReplica: 当一个replica变为一个parition的assingned replicas时
      其状态变为OnlineReplica, 即一个有效的OnlineReplica
      
      3 Online状态的parition才能转变为leader或isr中的一员
      
      4 OfflineReplica: 当一个broker down时, 上面的replica也随之die, 其状态转变为Onffline;
      ReplicaDeletionStarted: 当一个replica的删除操作开始时,其状态转变为ReplicaDeletionStarted
      
      5 ReplicaDeletionSuccessful: Replica成功删除后,其状态转变为ReplicaDeletionSuccessful
      
      6 ReplicaDeletionIneligible: Replica成功失败后,其状态转变为ReplicaDeletionIneligible
      
      7 NonExistentReplica:  Replica成功删除后, 从ReplicaDeletionSuccessful状态转变为NonExistentReplica状态
    复制代码
  • ReplicaStateMachine 所在文件: core/src/main/scala/kafka/controller/ReplicaStateMachine.scala

  • startup: 启动ReplicaStateMachine

  • initializeReplicaState: 初始化每个replica的状态, 如果replica所在的broker是live状态,则此replica的状态为OnlineReplica。

  • 处理可以转换到Online状态的Replica, handleStateChanges(controllerContext.allLiveReplicas(), OnlineReplica), 并且发送LeaderAndIsrRequest到各broker nodes: handleStateChanges(controllerContext.allLiveReplicas(), OnlineReplica)

  • 当创建某个topic时,该topic下所有分区的所有副本都是NonExistent。

  • 当controller加载Zookeeper中该topic每一个分区的所有副本信息到内存中,同时将副本的状态变更为New。

  • 之后controller选择该分区副本列表中的第一个副本作为分区的leader副本并设置所有副本进入ISR,然后在Zookeeper中持久化该决定。

  • 一旦确定了分区的Leader和ISR之后,controller会将这些消息以请求的方式发送给所有的副本。

  • 同时将这些副本状态同步到集群的所有broker上以便让他们知晓。

  • 最后controller 会把分区的所有副本状态设置为Online。

3 partitionStateMachine (根据副本分配方案创建分区)

  • Partition有如下四种状态

      NonExistentPartition: 这个partition还没有被创建或者是创建后又被删除了;
      NewPartition: 这个parition已创建, replicas也已分配好,但leader/isr还未就绪;
      OnlinePartition: 这个partition的leader选好;
      OfflinePartition: 这个partition的leader挂了,这个parition状态为OfflinePartition;
    复制代码
  • 当创建Topic时,controller负责创建分区对象,它首先会短暂的将所有分区状态设置为NonExistent。

  • 之后读取Zookeeper副本分配方案,然后令分区状态设置为NewPartion。

  • 处于NewPartion状态的分区尚未有leader和ISR,因此Controller会初始化leader和ISR信息并设置分区状态为OnlinePartion,此时分区正常工作。

  • 本套技术专栏是作者(秦凯新)平时工作的总结和升华,通过从真实商业环境抽取案例进行总结和分享,并给出商业应用的调优建议和集群环境容量规划等内容,请持续关注本套博客。期待加入IOT时代最具战斗力的团队。QQ邮箱地址:1120746959@qq.com,如有任何学术交流,可随时联系。

4 Controller职责所在(监听znode状态变化做执行)

  • UpdateMetadataRequest:更新元数据请求(比如:topic有多少个分区,每一个分区的leader在哪一台broker上以及分区的副本列表),随着集群的运行,这部分信息随时都可能变更,一旦发生变更,controller会将最新的元数据广播给所有存活的broker。具体方式就是给所有broker发送UpdateMetadataRequest请求
  • CreateTopics: 创建topic请求。当前不管是通过API方式、脚本方式(--create)抑或是CreateTopics请求方式来创建topic,做法几乎都是在Zookeeper的/brokers/topics下创建znode来触发创建逻辑,而controller会监听该path下的变更来执行真正的“创建topic”逻辑
  • DeleteTopics:删除topic请求。和CreateTopics类似,也是通过创建Zookeeper下的/admin/delete_topics/节点来触发删除topic,主要逻辑有:1:停止所有副本运行。2:删除所有副本的日志数据。3:移除zk上的 /admin/delete_topics/节点。
  • 分区重分配:即kafka-reassign-partitions脚本做的事情。同样是与Zookeeper结合使用,脚本写入/admin/reassign_partitions节点来触发,controller负责按照方案分配分区。执行过程是:先扩展再伸缩机制(就副本和新副本集合同时存在)。
  • Preferred leader分配:调整分区leader副本,preferred leader选举当前有两种触发方式:1. 自动触发(auto.leader.rebalance.enable = true),controller会自动调整Preferred leader。2. kafka-preferred-replica-election脚本触发。两者步骤相同,都是向Zookeeper的/admin/preferred_replica_election写数据,controller提取数据执行preferred leader分配
  • 分区扩展:即增加topic分区数。标准做法也是通过kafka-reassign-partitions脚本完成,不过用户可直接往Zookeeper中写数据来实现,比如直接把新增分区的副本集合写入到/brokers/topics/下,然后controller会为你自动地选出leader并增加分区
  • 集群扩展:新增broker时Zookeeper中/brokers/ids下会新增znode,controller自动完成服务发现的工作
  • broker崩溃:同样地,controller通过Zookeeper可实时侦测broker状态。一旦有broker挂掉了,controller可立即感知并为受影响分区选举新的leader
  • ControlledShutdown:broker除了崩溃,还能“优雅”地退出。broker一旦自行终止,controller会接收到一个ControlledShudownRequest请求,然后controller会妥善处理该请求并执行各种收尾工作
  • Controller leader选举:controller必然要提供自己的leader选举以防这个全局唯一的组件崩溃宕机导致服务中断。这个功能也是通过Zookeeper的帮助实现的。

5 Controller与Broker之间的通信机制(NIO select)

  • controller启动时会为集群中的所有Broker创建一个专属的Socket连接,加入有100台broker机器,那么controller会创建100个Socket连接。新版本目前统一使用NIO select ,实际上还是要维护100个线程。

6 ControllerContext数据组件

  • controller的缓存,可谓是最重要的数据组件了,ControllerContext汇总了Zookeeper中关于kafka集群中所有元数据信息,是controller能够正确提供服务的基础。

本套技术专栏是作者(秦凯新)平时工作的总结和升华,通过从真实商业环境抽取案例进行总结和分享,并给出商业应用的调优建议和集群环境容量规划等内容,请持续关注本套博客。期待加入IOT时代最具战斗力的团队。QQ邮箱地址:1120746959@qq.com,如有任何学术交流,可随时联系。

7 总结

kafka集群Controller主要干通过ZK持久化副本分配方案,根据副本分配方案创建分区,监听ZK znode状态变化做执行处理,维护分区和副本ISR机制稳定运行。感谢huxihx技术博客以及相关书籍,让我理解了Controller核心机制,写一篇学习笔记,作为总结,辛苦成文,实属不易,谢谢。

秦凯新 于深圳 201812021541

相关文章:

  • Linux C编程之一:Linux下c语言的开发环境
  • 写给高年级小学生看的《Bash 指南》
  • Windows10下 tensorflow-gpu 配置
  • 前端模板技术面面观(2)
  • 发布国内首个无服务器容器服务,运维效率从未如此高效
  • 操作系统-进程控制
  • 区块链在刚刚领域引起行业革新
  • pages bookmarks for machine learning domain
  • 2019年值得关注的八大DevOps趋势
  • JS 日期转换,格式化等常用的函数定义
  • AlphaZero进化论:从零开始,制霸所有棋类游戏
  • 一个程序员的自白(危机可导)
  • golang 学习笔记 ---数组/字符串/切片
  • 整数规划---割平面解法
  • 试水Proxmox最新版本PVE 5.3
  • [PHP内核探索]PHP中的哈希表
  • 《Java8实战》-第四章读书笔记(引入流Stream)
  • 【前端学习】-粗谈选择器
  • Android开发 - 掌握ConstraintLayout(四)创建基本约束
  • angular学习第一篇-----环境搭建
  • ES6 学习笔记(一)let,const和解构赋值
  • JavaSE小实践1:Java爬取斗图网站的所有表情包
  • Linux快速配置 VIM 实现语法高亮 补全 缩进等功能
  • Promise面试题2实现异步串行执行
  • python 学习笔记 - Queue Pipes,进程间通讯
  • RxJS: 简单入门
  • SSH 免密登录
  • Webpack 4 学习01(基础配置)
  • 高度不固定时垂直居中
  • 回顾 Swift 多平台移植进度 #2
  • 前端工程化(Gulp、Webpack)-webpack
  • 提醒我喝水chrome插件开发指南
  • 用Visual Studio开发以太坊智能合约
  • 3月7日云栖精选夜读 | RSA 2019安全大会:企业资产管理成行业新风向标,云上安全占绝对优势 ...
  • MyCAT水平分库
  • RDS-Mysql 物理备份恢复到本地数据库上
  • 关于Android全面屏虚拟导航栏的适配总结
  • 直播平台建设千万不要忘记流媒体服务器的存在 ...
  • # Java NIO(一)FileChannel
  • # 深度解析 Socket 与 WebSocket:原理、区别与应用
  • # 执行时间 统计mysql_一文说尽 MySQL 优化原理
  • #绘制圆心_R语言——绘制一个诚意满满的圆 祝你2021圆圆满满
  • (Redis使用系列) Springboot 实现Redis 同数据源动态切换db 八
  • (Redis使用系列) Springboot 使用redis实现接口Api限流 十
  • (八)光盘的挂载与解挂、挂载CentOS镜像、rpm安装软件详细学习笔记
  • (规划)24届春招和25届暑假实习路线准备规划
  • (正则)提取页面里的img标签
  • .h头文件 .lib动态链接库文件 .dll 动态链接库
  • .NET 8 中引入新的 IHostedLifecycleService 接口 实现定时任务
  • .NET Core实战项目之CMS 第十二章 开发篇-Dapper封装CURD及仓储代码生成器实现
  • .Net Redis的秒杀Dome和异步执行
  • .NET 分布式技术比较
  • .NET/C# 中设置当发生某个特定异常时进入断点(不借助 Visual Studio 的纯代码实现)
  • .ui文件相关
  • @31省区市高考时间表来了,祝考试成功