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

广州正规网站建设有哪些,福州招聘网站有哪几个,赣州营销公司,南冒网站建设制作推广公司第二章 Kafka设计原理详解 1、Kafka核心总控制器Controller 在 Kafka 集群中会有一个或者多个 broker#xff0c;其中有一个 broker 会被选举为控制器#xff08;Kafka Controller#xff09;#xff0c;它负责管理整个集群中所有分区和副本的状态。 当某个分区的 leader…第二章 Kafka设计原理详解 1、Kafka核心总控制器Controller 在 Kafka 集群中会有一个或者多个 broker其中有一个 broker 会被选举为控制器Kafka Controller它负责管理整个集群中所有分区和副本的状态。 当某个分区的 leader 副本出现故障时由控制器负责为该分区选举新的 leader 副本。当检测到某个分区的 ISR 集合发生变化时由控制器负责通知所有 broker 更新其元数据信息。 当使用 kafka-topics.sh 脚本为某个 topic 增加分区数量时同样还是由控制器负责让新分区被其他节点感知到。 2、Controller选举机制 在 kafka 集群启动的时候会自动选举一台 broker 作为 controller 来管理整个集群选举的过程是集群中每个 broker 都会尝试在 zookeeper 上创建一个 /controller 临时节点zookeeper 会保证有且仅有一个 broker 能创建成功这个 broker 就会成为集群的总控器 controller。 当这个 controller 角色的 broker 宕机了此时 zookeeper 临时节点/controller会消失集群里的其他 broker 会一直监听这个临时节点发现临时节点消失了就再次竞争创建临时节点就是我们上面说的选举机制zookeeper 又会保证有一个 broker 成为新的 controller。 具备控制器身份的 broker 需要比其他普通的 broker 多一份职责具体细节如下 1、监听 broker 相关的变化。为 Zookeeper 中的 /brokers/ids 节点添加 BrokerChangeListener用来处理 broker 增减的变化。 2、监听 topic 相关的变化。为 Zookeeper 中的 /brokers/topics 节点添加 TopicChangeListener用来处理 topic 增减的变化为 Zookeeper 中的 /admin/delete_topics 节点添加 TopicDeletionListener用来处理删除 topic 的动作。 3、从 Zookeeper 中获取当前所有与 topic、partition 以及 broker 有关的信息并进行相应的管理。对于所有 topic 所对应的 Zookeeper 中的 /brokers/topics/[topic] 节点添加 PartitionModificationsListener用来监听 topic 中的分区分配变化。 4、更新集群的元数据信息同步到其他普通的 broker 节点中。 3、Partition副本选举Leader机制 controller 感知到分区副本 leader 所在的 broker 挂了controller监听了很多 zk 节点可以感知到 broker 存活controller 会从 ISR 列表参数unclean.leader.election.enablefalse 的前提下里挑第一个 broker 作为 leader第一个 broker 最先放进 ISR 列表可能是同步数据最多的副本如果参数 unclean.leader.election.enable 为 true代表在 ISR 列表里所有副本都挂了的时候可以在 ISR 列表以外的副本中选 leader这种设置可以提高可用性但是选出的新 leader 有可能数据少很多。 副本进入 ISR 列表有两个条件 1、副本节点不能产生分区必须能与 zookeeper 保持会话以及跟 leader 副本网络连通 2、副本能复制 leader 上的所有写操作并且不能落后太多。与 leader 副本同步滞后的副本是由 replica.lag.time.max.ms 配置决定的超过这个时间都没有跟 leader 副本同步过一次的副本会被移出 ISR 列表 4、消费者消费消息的offset记录机制 每个 consumer 会定期将自己消费分区的 offset 提交给 kafka 内部 topic__consumer_offsets提交过去的时候key 是 consumerGroupIdtopic分区号value 就是当前 offset 的值kafka 会定期清理 topic 里的消息最后就保留最新的那条数据 因为 __consumer_offsets 可能会接收高并发的请求kafka 默认给其分配 50 个分区可以通过 offsets.topic.num.partitions 设置这样可以通过加机器的方式扩大并发。 通过如下公式可以选出 consumer 消费的 offset 要提交到 __consumer_offsets 的哪个分区 公式hash(consumerGroupId) % __consumer_offsets 的分区数 5、消费者Rebalance机制 rebalance 就是说如果消费组里的消费者数量有变化或消费的分区数有变化时kafka 会重新分配消费者消费分区的关系。比如 consumer group 中的某个消费者挂了此时会自动把分配给他的分区交给其他的消费者如果他又重启了那么又会把一些分区重新交还给他。 注意rebalance 只针对 subscribe 这种不指定分区消费的情况如果通过 assign 这种消费方式指定了分区kafka 不会进行 rebanlance。 如下情况可能会触发消费者 rebalance 1、消费组里的 consumer 增加或减少了 2、动态给 topic 增加了分区 3、消费组订阅了更多的 topic rebalance 过程中消费者无法从 kafka 消费消息这对 kafka 的 TPS 会有影响如果 kafka 集群内节点较多比如数百个那重平衡可能会耗时极多所以应尽量避免在系统高峰期发生重平衡。 5.1、消费者Rebalance分区分配策略 主要有三种 rebalance 策略range、round-robin、sticky。 Kafka 提供了消费者客户端参数 partition.assignment.strategy 来设置消费者与订阅主题之间的分区分配策略。默认情况为 range 分配策略。 假设一个主题有 10 个分区0-9现在有三个 consumer 消费 1、range 策略就是按照分区序号排序假设 n分区数消费者数量 3 m分区数 % 消费者数量 1那么前 m 个消费者每个分配 n1 个分区后面的消费者数量m 个消费者每个分配 n 个分区。 比如分区 0~3 给一个 consumer分区 4~6 给一个 consumer分区 7~9 给一个 consumer。 2、round-robin 策略就是轮询分配比如分区 0、3、6、9 给一个 consumer分区 1、4、7 给一个 consumer分区 2、5、8 给一个 consumer 3、sticky 策略初始时分配策略与 round-robin 类似但是在 rebalance 的时候需要保证如下两个原则 分区的分配要尽可能均匀分区的分配尽可能与上次分配保持相同 当两者发生冲突时第一个目标优先于第二个目标 。这样可以最大程度维持原来的分区分配策略。 比如对于第一种 range 情况的分配如果第三个 consumer 挂了那么重新用 sticky 策略分配的结果如下 consumer1 除了原有的 0~3会再分配一个 7 consumer2 除了原有的 4~6会再分配 8 和 9 5.2、Rebalance过程 当有消费者加入消费组时消费者、消费组以及组协调器之间会经历以下几个阶段 第一阶段选择组协调器 组协调器 GroupCoordinator每个 consumer group 都会选择一个 broker 作为自己的组协调器 coordinator负责监控这个消费组里的所有消费者的心跳以及判断是否宕机然后开启消费者 rebalance。 consumer group 中的每个 consumer 启动时会向 kafka 集群中的某个节点发送 FindCoordinatorRequest 请求来查找对应的组协调器 GroupCoordinator并跟其建立网络连接。 组协调器选择方式 consumer 消费的 offset 要提交到 __consumer_offsets 的哪个分区这个分区 leader 对应的 broker 就是这个 consumer group 的 coordinator 第二阶段加入消费组 JOIN GROUP 在成功找到消费组所对应的 GroupCoordinator 之后就进入加入消费组的阶段在此阶段的消费者会向 GroupCoordinator 发送 JoinGroupRequest 请求并处理响应。然后 GroupCoordinator 从一个 consumer group 中选择第一个加入 group 的 consumer 作为 leader消费组协调器把 consumer group 情况发送给这个 leader接着这个 leader 会负责制定分区方案。 第三阶段SYNC GROUP consumer leader 通过给 GroupCoordinator 发送 SyncGroupRequest接着 GroupCoordinator 就把分区方案下发给各个 consumer他们会根据指定分区的leader broker 进行网络连接以及消息消费。 6、日志分段存储 Kafka 一个分区的消息数据对应存储在一个文件夹下以 topic 名称 分区号命名消息在分区内是分段segment存储的每个分段的消息都存储在不同的 log 文件里这种特性方便 old segment file 快速被删除kafka 规定了一个段位的 log 文件最大为 1G做这个限制的目的是为了方便把 log 文件加载到内存去操作

部分消息的offset索引文件kafka每次往分区发4K(可配置)消息就会记录一条当前消息的offset到index文件

如果要定位消息的offset会先在这个文件里快速定位再去log文件里找具体消息

00000000000000000000.index

消息存储文件主要存offset和消息体

00000000000000000000.log

消息的发送时间索引文件kafka每次往分区发4K(可配置)消息就会记录一条当前消息的发送时间戳与对应的offset到timeindex文件

如果需要按照时间来定位消息的offset会先在这个文件里查找

00000000000000000000.timeindex00000000000005367851.index 00000000000005367851.log 00000000000005367851.timeindex00000000000009936472.index 00000000000009936472.log 00000000000009936472.timeindex这个 9936472 之类的数字就是代表了这个日志段文件里包含的起始 Offset也就说明这个分区里至少都写入了接近 1000 万条数据了。 Kafka Broker 有一个参数log.segment.bytes限定了每个日志段文件的大小最大就是 1GB。 一个日志段文件满了就自动开一个新的日志段文件来写入避免单个文件过大影响文件的读写性能这个过程叫做 log rolling正在被写入的那个日志段文件叫做 active log segment。