聊聊kafka的group coordinator

2017-10-03 本文已影响1853人 go4it

序

本文主要来讲一个kafka的group coordinator。在kafka0.9.0版本的时候，开始启用了新的consumer config，这个新的consumer config采用bootstrap.servers替代之前版本的zookeeper.connect，主要是要渐渐弱化zk的依赖，把zk依赖隐藏到broker背后。

group coordinator

使用bootstrap.servers替代之前版本的zookeeper.connect，相关的有如下两个改动：

在 Server 端增加了 GroupCoordinator 这个角色
将 topic 的 offset 信息由之前存储在 zookeeper(/consumers/<group.id>/offsets/<topic>/<partitionId>,zk写操作性能不高) 上改为存储到一个特殊的 topic 中（__consumer_offsets）

从0.8.2版本开始Kafka开始支持将consumer的位移信息保存在Kafka内部的topic中（从0.9.0版本开始默认将offset存储到系统topic中）

Coordinator一般指的是运行在broker上的group Coordinator，用于管理Consumer Group中各个成员，每个KafkaServer都有一个GroupCoordinator实例，管理多个消费者组，主要用于offset位移管理和Consumer Rebalance。

rebalance时机

在如下条件下，partition要在consumer中重新分配：

条件1：有新的consumer加入
条件2：旧的consumer挂了
条件3：coordinator挂了，集群选举出新的coordinator
条件4：topic的partition新加
条件5：consumer调用unsubscrible()，取消topic的订阅

__consumer_offsets

Consumer通过发送OffsetCommitRequest请求到指定broker（偏移量管理者）提交偏移量。这个请求中包含一系列分区以及在这些分区中的消费位置（偏移量）。偏移量管理者会追加键值（key－value）形式的消息到一个指定的topic（__consumer_offsets）。key是由consumerGroup-topic-partition组成的，而value是偏移量。

offset1.png

内存中也会维护一份最近的记录，为了在指定key的情况下能快速的给出OffsetFetchRequests而不用扫描全部偏移量topic日志。如果偏移量管理者因某种原因失败，新的broker将会成为偏移量管理者并且通过扫描偏移量topic来重新生成偏移量缓存。

屏幕快照 2017-10-02 下午9.21.00.png

清除offset日志

配置

log.cleaner.enable=true

compact

屏幕快照 2017-10-02 下午9.26.19.png