Coordinator与offset管理和Consumer Rebalance

谁践踏了优雅 2023-01-19 13:49 7阅读 0赞

## 1、Consumer与Consumer Group ##

consumer group是kafka提供的可扩展且具有容错性的消费者机制。组内可以有多个消费者或消费者实例(consumer instance)，它们共享一个公共的ID，即group ID。组内的所有消费者协调在一起来消费订阅主题(subscribed topics)的所有分区(partition)。  
consumer group下可以有一个或多个consumer instance，consumer instance可以是一个进程，也可以是一个线程  
group.id是一个字符串，唯一标识一个consumer group  
consumer group下订阅的topic下的每个分区只能分配给某个group下的一个consumer(当然该分区还可以被分配给其他group)

![format_png][]

图片.png

## 2、Coordinator介绍 ##

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

# Coordinator存储的信息 #

对于每个Consumer Group，Coordinator会存储以下信息：

1.  对每个存在的topic，可以有多个消费组group订阅同一个topic(对应消息系统中的广播)
2.  对每个Consumer Group，元数据如下：  
    订阅的topics列表  
    Consumer Group配置信息，包括session timeout等  
    组中每个Consumer的元数据。包括主机名，consumer id  
    每个正在消费的topic partition的当前offsets  
    Partition的ownership元数据，包括consumer消费的partitions映射关系

# 如何确定consumer group的coordinator #

consumer group如何确定自己的coordinator是谁呢？ 简单来说分为两步：

1.  确定consumer group位移信息写入\_\_consumers\_offsets这个topic的哪个分区。具体计算公式：  
    \_\_consumers\_offsets partition\# = Math.abs(groupId.hashCode() % groupMetadataTopicPartitionCount) 注意：groupMetadataTopicPartitionCount由offsets.topic.num.partitions指定，默认是50个分区。
2.  该分区leader所在的broker就是被选定的coordinator

# Coordinator工作过程 #

## 3、offset位移管理 ##

消费者在消费的过程中需要记录自己消费了多少数据，即消费位置信息。在Kafka中这个位置信息有个专门的术语：位移(offset)。  
(1)、很多消息引擎都把这部分信息保存在服务器端(broker端)。这样做的好处当然是实现简单，但会有三个主要的问题：  
1. broker从此变成有状态的，会影响伸缩性；  
2. 需要引入应答机制(acknowledgement)来确认消费成功。  
3. 由于要保存很多consumer的offset信息，必然引入复杂的数据结构，造成资源浪费。  
而Kafka选择了不同的方式：每个consumer group管理自己的位移信息，那么只需要简单的一个整数表示位置就够了；同时可以引入checkpoint机制定期持久化，简化了应答机制的实现。  
(2)、Kafka默认是定期帮你自动提交位移的(enable.auto.commit = true)，你当然可以选择手动提交位移实现自己控制。  
(3)、另外kafka会定期把group消费情况保存起来，做成一个offset map，如下图所示：

![format_png 1][]

图片.png

上图中表明了test-group这个组当前的消费情况。

# offset管理 #

老版本的位移是提交到zookeeper中的，目录结构是：/consumers/<group.id>/offsets/<topic>/<partitionId>，但是zookeeper其实并不适合进行大批量的读写操作，尤其是写操作。

![format_png 2][]

图片.png

因此kafka提供了另一种解决方案：增加**\_\_consumeroffsets topic**，将offset信息写入这个topic，摆脱对zookeeper的依赖(指保存offset这件事情)。\_\_consumer\_offsets中的消息保存了每个consumer group某一时刻提交的offset信息。依然以上图中的consumer group为例，格式大概如下：

![format_png 3][]

图片.png

![format_png 4][]

图片.png

![format_png 5][]

图片.png

\_\_consumers\_offsets topic配置了compact策略，使得它总是能够保存最新的位移信息，既控制了该topic总体的日志容量，也能实现保存最新offset的目的。compact的具体原理请参见：[Log Compaction][]  
至于每个group保存到\_\_consumers\_offsets的哪个分区，如何查看的问题请参见这篇文章：[Kafka 如何读取offset topic内容 (\_\_consumer\_offsets)][Kafka _offset topic_ _consumer_offsets]

# 位移管理的过程 #

offset提交消息会根据消费组的key(消费组名称)进行分区. 对于一个给定的消费组,它的所有消息都会发送到唯一的broker(即Coordinator)  
Coordinator上负责管理offset的组件是**Offset manager**。负责存储,抓取,和维护消费者的offsets. 每个broker都有一个offset manager实例. 有两种具体的实现:  
ZookeeperOffsetManager: 调用zookeeper来存储和接收offset（老版本的位移管理）。  
DefaultOffsetManager: 提供消费者offsets内置的offset管理。  
通过在config/server.properties中的offset.storage参数选择。  
**DefaultOffsetManager**  
除了将offset作为logs保存到磁盘上,DefaultOffsetManager维护了一张能快速服务于offset抓取请求的**consumer offsets表**。这个表作为缓存，包含的含仅仅是”offsets topic”的partitions中属于leader partition对应的条目(存储的是offset)。  
对于DefaultOffsetManager还有两个其他属性: “offsets.topic.replication.factor和”offsets.topic.num.partitions”,默认值都是1。这两个属性会用来自动地创建”offsets topic”。  
**offset manager接口的概要：**

![format_png 6][]

图片.png

**Offset Commit提交过程：**  
**消费端**  
一条offset提交消息会作为生产请求.当消费者启动时,会为”offsets topic”创建一个消费者. 下面是内置的生产者的一些属性:  
可以使用异步.但是使用同步可以避免延迟的生产请求(因为是批量消息),并且我们需要立即知道offset消息是否被broker成功接收  
|request.required.acks|-1|确保所有的replicas和leader是同步的,并且能看到所有的offset消息  
|key.serializer.class|StringEncoder|key和payload都是strings  
注意我们没有对提交的offset消息进行压缩,因为每条消息本身大小是很小的,如果压缩了反而适得其反.  
目前key和offset的值通过纯文本方式传递. 我们可以转换为更加紧凑的二进制协议,而不是把  
Long类型的offset和Int类型的partition作为字符串. 当然在不断演进时还要考虑版本和格式协议.  
**broker端**  
broker把接收到的offset提交信息当做一个正常的生产请求,对offset请求的处理和正常的生产者请求处理方式是一样的.  
一旦将数据追加到leader的本地日志中,并且所有的replicas都赶上leader.leader检查生产请求是”offsets topic”,  
(因为broker端的处理逻辑针对offset请求和普通生产请求是一样的,如果是offset请求,还需要有不同的处理分支)  
它就会要求offset manager添加这个offset(对于延迟的生产请求,更新操作会在延迟的生产请求被完成的时候).  
因为设置了acks=-1,只有当这些offsets成功地复制到ISR中的所有brokers,才会被提交给offset manager.

![format_png 7][]

图片.png

**Offset Fetch获取过程：**  
**消费端**  
消费者启动时,会首先创建到任意一个存活的brokers的通道.因此消费者会发送它所有”OffsetFetchRequest”  
到这个随机选中的broker. 如果出现错误,这个通道就会被关闭,并重新创建一个随机的通道.  
**broker端**  
一个Offset抓取请求包含了多个topic-partitions. 接收请求的broker可能有也可能没有请求的partitions的offset信息.  
因此接收请求的brokers也会和其他broker通信. 一个通道连接池会用来转发请求给partition的leader broker.  
下面是一个broker在接收到一个offset抓取请求后的步骤:  
接收请求的broker首先决定”offset topic”的哪个partition负责这个请求  
从broker的leader cache中找出对应partition的leader(会在controller的每次metadata更新请求中更新缓存)  
如果接收请求的broker就是leader,它会从自己的offset manager中读取出offset,并添加到响应中  
如果offset不存在,返回UnknownTopicOrPartitionCode  
如果broker正在加载offsets table,返回OffsetLoadingCode.消费者受到这个状态码会在之后重试  
如果接收请求的broker不是指定topic-partition的leader,它会将OffsetFetchRequest转发给这个partition的当前leader  
如果”offsets topic”这个时候不存在,它会尝试自动创建,在创建成果后,会返回offset=-1

![format_png 8][]

图片.png

## 3、Consumer Reblance ##

**什么是rebalance？**  
rebalance本质上是一种协议，规定了一个consumer group下的所有consumer如何达成一致来分配订阅topic的每个分区。比如某个group下有20个consumer，它订阅了一个具有100个分区的topic。正常情况下，Kafka平均会为每个consumer分配5个分区。这个分配的过程就叫rebalance。Kafka新版本consumer默认提供了两种分配策略：range和round-robin。  
**rebalance的触发条件有三种：**  
组成员发生变更(新consumer加入组、已有consumer主动离开组或已有consumer崩溃了——这两者的区别后面会谈到)  
订阅主题数发生变更——这当然是可能的，如果你使用了正则表达式的方式进行订阅，那么新建匹配正则表达式的topic就会触发rebalance  
订阅主题的分区数发生变更

![format_png 9][]

图片.png

**Rebalance Generation**  
JVM GC的分代收集就是这个词(严格来说是generational)，我这里把它翻译成“届”好了，它表示了rebalance之后的一届成员，主要是用于保护consumer group，隔离无效offset提交的。比如上一届的consumer成员是无法提交位移到新一届的consumer group中。我们有时候可以看到ILLEGAL\_GENERATION的错误，就是kafka在抱怨这件事情。每次group进行rebalance之后，generation号都会加1，表示group进入到了一个新的版本，如下图所示： Generation 1时group有3个成员，随后成员2退出组，coordinator触发rebalance，consumer group进入Generation 2，之后成员4加入，再次触发rebalance，group进入Generation 3.

![format_png 10][]

图片.png

**协议(protocol)**  
rebalance本质上是一组协议。group与coordinator共同使用它来完成group的rebalance。目前kafka提供了5个协议来处理与consumer group coordination相关的问题：  
Heartbeat请求：consumer需要定期给coordinator发送心跳来表明自己还活着  
LeaveGroup请求：主动告诉coordinator我要离开consumer group  
SyncGroup请求：group leader把分配方案告诉组内所有成员  
JoinGroup请求：成员请求加入组  
DescribeGroup请求：显示组的所有信息，包括成员信息，协议名称，分配方案，订阅信息等。通常该请求是给管理员使用  
Coordinator在rebalance的时候主要用到了前面4种请求。  
**liveness**  
consumer如何向coordinator证明自己还活着？ 通过定时向coordinator发送Heartbeat请求。如果超过了设定的超时时间，那么coordinator就认为这个consumer已经挂了。一旦coordinator认为某个consumer挂了，那么它就会开启新一轮rebalance，并且在当前其他consumer的心跳response中添加“REBALANCE\_IN\_PROGRESS”，告诉其他consumer：不好意思各位，你们重新申请加入组吧！  
**Rebalance过程**  
rebalance的前提是coordinator已经确定了。  
总体而言，rebalance分为2步：Join和Sync  
1 Join， 顾名思义就是加入组。这一步中，所有成员都向coordinator发送JoinGroup请求，请求入组。一旦所有成员都发送了JoinGroup请求，coordinator会从中选择一个consumer担任leader的角色，并把组成员信息以及订阅信息发给leader——注意leader和coordinator不是一个概念。leader负责消费分配方案的制定。

![format_png 11][]

Paste\_Image.png

2 Sync，这一步leader开始分配消费方案，即哪个consumer负责消费哪些topic的哪些partition。一旦完成分配，leader会将这个方案封装进SyncGroup请求中发给coordinator，非leader也会发SyncGroup请求，只是内容为空。coordinator接收到分配方案之后会把方案塞进SyncGroup的response中发给各个consumer。这样组内的所有成员就都知道自己应该消费哪些分区了。

![format_png 12][]

Paste\_Image.png

**注意！！ consumer group的分区分配方案是在客户端执行的！**Kafka将这个权利下放给客户端主要是因为这样做可以有更好的灵活性。比如这种机制下我可以实现类似于Hadoop那样的机架感知(rack-aware)分配方案，即为consumer挑选同一个机架下的分区数据，减少网络传输的开销。Kafka默认为你提供了两种分配策略：range和round-robin。由于这不是本文的重点，这里就不再详细展开了，你只需要记住你可以覆盖consumer的参数：partition.assignment.strategy来实现自己分配策略就好了。  
**consumer group状态机**  
和很多kafka组件一样，group也做了个状态机来表明组状态的流转。coordinator根据这个状态机会对consumer group做不同的处理，如下图所示

![format_png 13][]

Paste\_Image.png

简单说明下图中的各个状态：  
Dead：组内已经没有任何成员的最终状态，组的元数据也已经被coordinator移除了。这种状态响应各种请求都是一个response： UNKNOWN\_MEMBER\_ID  
Empty：组内无成员，但是位移信息还没有过期。这种状态只能响应JoinGroup请求  
PreparingRebalance：组准备开启新的rebalance，等待成员加入  
AwaitingSync：正在等待leader consumer将分配方案传给各个成员  
Stable：rebalance完成！可以开始消费了  
**rebalance场景剖析**  
1 新成员加入组(member join)

![format_png 14][]

Paste\_Image.png

2 组成员崩溃(member failure)  
**组成员崩溃和组成员主动离开是两个不同的场景。**因为在崩溃时成员并不会主动地告知coordinator此事，coordinator有可能需要一个完整的session.timeout周期(心跳周期)才能检测到这种崩溃，这必然会造成consumer的滞后。可以说离开组是主动地发起rebalance；而崩溃则是被动地发起rebalance。如图：

![format_png 15][]

Paste\_Image.png

3 组成员主动离组（member leave group)

![format_png 16][]

Paste\_Image.png

4 提交位移(member commit offset)

![format_png 17][]

Paste\_Image.png

refer  
[http://www.cnblogs.com/huxi2b/p/6223228.html][http_www.cnblogs.com_huxi2b_p_6223228.html]  
[http://zqhxuyuan.github.io/2016/02/18/Kafka-Consumer-Offset-Manager/][http_zqhxuyuan.github.io_2016_02_18_Kafka-Consumer-Offset-Manager]  
[http://www.cnblogs.com/byrhuangqiang/p/6384986.html][http_www.cnblogs.com_byrhuangqiang_p_6384986.html]

作者：康康不遛猫  
链接：https://www.jianshu.com/p/5aa8776868bb/  
来源：简书  
简书著作权归作者所有，任何形式的转载都请联系作者获得授权并注明出处。

[format_png]: /images/20221021/e5d69dddcb0647df861d2fd163064141.png
[format_png 1]: /images/20221021/9398c272c8c5494695377706acbce04d.png
[format_png 2]: /images/20221021/6610d29c44fd4784a432c467c8c992fe.png
[format_png 3]: /images/20221021/ef57f46c8ffb4a63920066e56e50b616.png
[format_png 4]: /images/20221021/3632faf126114fa4937238a902be439b.png
[format_png 5]: /images/20221021/2f68c701549e41a5ba5bd750d96dd1ec.png
[Log Compaction]: https://link.jianshu.com?t=https://kafka.apache.org/documentation/#compaction
[Kafka _offset topic_ _consumer_offsets]: https://link.jianshu.com?t=http://www.cnblogs.com/huxi2b/p/6061110.html
[format_png 6]: /images/20221021/fde557138c794d13a7bb1d3cb30be3ef.png
[format_png 7]: /images/20221021/2f60ff7c9f8a412e92d06b2837b30243.png
[format_png 8]: /images/20221021/04815a1c6d82448db32fa95268a1781d.png
[format_png 9]: /images/20221021/95d500b2420c4b4d98c480e0599ed790.png
[format_png 10]: /images/20221021/42cc14ea5d0b404bb316d83a8def4d9f.png
[format_png 11]: /images/20221021/f82a591d64064d87aae481f1b6a11b61.png
[format_png 12]: /images/20221021/3c563b876f4d4756b9e5d8530db1b566.png
[format_png 13]: /images/20221021/3da501bbbb364b64bd86d0df629a9891.png
[format_png 14]: /images/20221021/76f5b7345cf4457084b1f604a1b16b23.png
[format_png 15]: /images/20221021/ec558375450645fb97995abca92b9d3e.png
[format_png 16]: /images/20221021/cb155ccc5d6d42e7abb82b3074e7b446.png
[format_png 17]: /images/20221021/4340cada7cb24bd694d28a1dfdf97441.png
[http_www.cnblogs.com_huxi2b_p_6223228.html]: https://link.jianshu.com?t=http://www.cnblogs.com/huxi2b/p/6223228.html
[http_zqhxuyuan.github.io_2016_02_18_Kafka-Consumer-Offset-Manager]: https://link.jianshu.com?t=http://zqhxuyuan.github.io/2016/02/18/Kafka-Consumer-Offset-Manager/
[http_www.cnblogs.com_byrhuangqiang_p_6384986.html]: https://link.jianshu.com?t=http://www.cnblogs.com/byrhuangqiang/p/6384986.html