【Kafka】原理分析：Consumer消费规则

￡神魔★判官ぃ 2022-12-21 04:54 144阅读 0赞

在实际生产过程中，每个topic都会有多个partitions，多个partitions的好处在于，一方面能够对 broker上的数据进行分片有效减少了消息的容量从而提升io性能。另外一方面，为了提高消费端的消费 能力，一般会通过多个consumer去消费同一个topic ，也就是消费端的负载均衡机制。这也就是我们接下来要了解的，在多个partition以及多个consumer的情况下，消费者是如何消费消息的。

同时，在[【Kafka】基本使用：关于Consumer的几个概念][Kafka_Consumer]我们说了kafka存在consumer group的概念，kafka存在consumer group的概念，也就是group.id一样的 consumer，这些consumer属于一个consumer group，组内的所有消费者协调在一起来消费订阅主题的所有分区。

同一个partition，只能被消费组里的一个消费者消费，但是可以同时被多个消费组消费。那么同一个consumer group里面的consumer是怎么去分配该消费哪个分区里的数据的呢？如下图所示，3个分区，3个消费者，那么哪个消费者消分哪个分区？  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]  
对于上面这个图来说，这3个消费者会分别消费test这个topic 的3个分区，也就是每个consumer消费一个partition。

## 1.分区数量规则（宏观） ##

在分配partition时，首先要看的就是partition数量与consumer数量关系：

*  partition = consumer（3个partiton --> 3个consumer）：一对一
 *  partition > consumer（3个partiton --> 2个consumer）
    
     *  consumer1会消费partition0/partition1分区
     *  consumer2会消费partition2分区
 *  partition < consumer（3个partition --> 4个或以上consumer）
    
     *  仍然只有3个consumer对应3个partition
     *  其他的consumer无法消费消息

> consumer和partition数量建议：
> 
> 1.  如果consumer比partition多，是浪费。因为kafka的设计是在一个partition上是不允许并发的， 所以consumer数不要大于partition数
> 2.  如果consumer比partition少，一个consumer会对应于多个partitions，这里主要合理分配 consumer数和partition数，否则会导致partition里面的数据被取的不均匀。  
>     当consumer从多个partition读到数据，不保证数据间的顺序性，kafka只保证在一个partition 上数据是有序的，但多个partition，根据读的顺序会有不同
> 
> ==> Partitions : Consumers = 1 : 1是最好的，再不行也要保证partiton数目是consumer数目的整数倍，比如3c去消费12p。

## 2.分区分配策略（微观） ##

分区分配策略就是每个consumer具体消费哪个partition。在kafka中，存在三种分区分配策略，一种是Range(默认)、 一种是RoundRobin（轮询）、 一种是StickyAssignor(粘性)。

在消费端中的ConsumerConﬁg中，通过这个属性来指定分区分配策略

public static final String PARTITION_ASSIGNMENT_STRATEGY_CONFIG = "partition.assignment.strategy";

### 2.1 RangeAssignor（范围分区） ###

范围分区：Range策略是对每个主题而言的，首先对同一个主题里面的分区按照序号进行排序，并对消费者按照字母顺序进行排序。

n = 分区数／消费者数量  // 代表每个consumer基本消费分区数
    m = 分区数％消费者数量  // 代表剩余分区数 
    ==> 那么前m个消费者每个分配n+l个分区，后面的（消费者数量-m)个消费者每个分配n个分区

示例一：1个topic，10个分区，3个消费者

1.  排完序的分区将会是0, 1, 2, 3, 4, 5, 6, 7, 8, 9；消费者线程排完序将会是C1 C2, C3。
2.  n = 10 / 3 = 3，m = 10%3=1，那么消费者线程 C1-0 将会多消费一个分区

C1 将消费 0, 1, 2, 3 分区 
    C2 将消费 4, 5, 6 分区 
    C3 将消费 7, 8, 9 分区

示例二：2个topic（T1,T2），各10个分区，3个消费者

C1 将消费 T1主题的 0, 1, 2, 3 分区以及 T2主题的 0, 1, 2, 3分区 
    C2 将消费 T1主题的 4, 5, 6 分区以及 T2主题的 4, 5, 6分区 
    C3 将消费 T1主题的 7, 8, 9 分区以及 T2主题的 7, 8, 9分区

可以看出，C1-0 消费者线程比其他消费者线程多消费了2个分区，这就是Range strategy的一个很明显的弊端

### 2.2 RoundRobinAssignor（轮询分区） ###

轮询分区：把所有partition和所有consumer线程都列出来，然后按照hashcode进行排序。最后通过轮询算法分配partition给消费线程。如果所有consumer实例的订阅是相同的，那么partition会均匀分布。

示例：1个topic（T1），10个分区，3个消费者

1.  分区名按照 hashCode 排序：T1-5, T1-3, T1-0, T1-8, T-2, T1-1, T1-4, T1-7, T1-6, T1-9
2.  消费者线程排序：C1, C2，C3
3.  依次轮询分配

C1 将消费 T1-5, T1-8, T1-4, T1-9 分区； 
    C1 将消费 T1-3, T1-2, T1-7 分区；
    C3 将消费 T1-0, T1-1, T1-6 分区；

这也就意味着只要不触发动态平衡，C1就会固定消费5/8/4/9分区。

### 2.3 StrickyAssignor（粘滞策略） ###

粘滞策略：它主要有两个目的，当两者发生冲突时， 第一个目标优先于第二个目标。

1.  分区的分配尽可能的均匀
2.  分区的分配尽可能和上次分配保持相同

鉴于这两个目标， StickyAssignor分配策略的具体实现要比RangeAssignor和RoundRobinAssignor这两种分配策略要复杂得多。

示例：4个topic（t0,t1,t2,t3)，各2个分区（p0,p1），3个消费者（C0,C1,C2）；即一共8个分区

t0p0 、 t0p1 、 t1p0 、 t1p1 、 t2p0 、 t2p1 、t3p0 、 t3p1

最终分配结果为：

CO: tOpO、tlpl 、t3p0
    Cl: tOpl、t2p0 、t3pl 
    C2: tlpO、t2pl

上面看着有些像轮询，他俩的区别就在于Rebalance时两者的思想不同，strickyAssignor它是一种粘滞策略，所以它会满足**分区的分配尽可能和上次分配保持相同**。比如在某一时刻C1挂了，所以要重新分区（rebalance），根据strickyAssignor分配的结果应该是：

CO: tOpO、tlpl、t3p0、   t2p0    // 这正是与轮询的不同，轮询会将C1，C2已有的分区都进行重新分配
    C2: tlpO、t2pl、  tOpl、t3pl     // 给C2分配了两个分区

这种策略的好处是使得分区发生变化时，由于分区的“粘性，减少了不必要的分区移动。

## 3.动态平衡 （落实） ##

上面我们看了分区数量规则和分区分配策略，但如果消费者数量或者分区数量产生了变化，打破了现有的平衡，那该怎么办呢？

*  consumer数量变化:
    
     *  同一个consumer group内新增了消费者
     *  消费者离开当前所属的consumer group（比如主动停机或者宕机）
 *  partition数量变化：topic新增了分区

方案一：静态设置，直接指定consumer消费哪个分区

//消费指定分区的时候，不需要再订阅 
    //kafkaConsumer.subscribe(Collections.singletonList(topic)); 
    //消费指定的分区 
    TopicPartition topicPartition=new TopicPartition(topic,0); 
    kafkaConsumer.assign(Arrays.asList(topicPartition));

弊端：不符合高可用，当c-p其中任一方发生变化时，需等到重新手动配置后，整体才能恢复正常（有一段异常期）

方案二：**动态平衡**，在consumer或partition发生变化时动态平衡（也称为rebalance）。

> 既然是动态平衡，那肯定得有人去做啊，所以到底是由谁去做这项工作的呢？  
> 答：**coordinator**：具体执行rebalance和group管理。  
> 对于每个consumer group子集，都会在服务端对应一个Coordinator进行管理， Coordinator会在zookeeper上添加watcher，当消费者加入或者退出consumer group时，会修改zookeeper上保存的数据，从而触发GroupCoordinator开始Rebalance操作

整个rebalance的过程分为两个步骤， Join和Sync，但在rebalance之前，需要保证coordinator是已经确定好了的。

### 3.1 确定coordinator ###

当消费者准备加入某个Consumer group或者Coordinator发生故障转移时，消费者并不知道 GroupCoordinator在网络中的位置，这个时候就需要确定Coordinator。

消费者会向集群中任意一个Broker节点发送ConsumerMetadataRequest请求，收到请求的broker会返回一个response 作为响应，其中包含管理当前ConsumerGroup的GroupCoordinator。

> consumer group如何确定自己的coordinator是谁呢？服务端会返回一个负载最小的broker节点的id，并将该broker设置为coordinator。

消费者会根据broker的返回信息，连接到Coordinator，并且发送HeartbeatRequest（心跳），发送心跳的目的是要告诉Coordinator消费者(我）正常在线。当消费者在指定时间内没有发送心跳请求，则Coordinator会触发Rebalance操作。

### 3.2 JoinGroup ###

join: 表示加入到consumer group中。在这一步的过程如下：

1).**所有**的成员都会向coordinator发送joinGroup的请求。

2).所有成员都发送了joinGroup请求后，coordinator会选择一个consumer担任leader角色

leader选举算法比较简单，如果消费组内没有leader，那么第一个加入消费组的消费者就是消费者 leader，如果这个时候leader消费者退出了消费组，那么重新选举一个leader，这个选举很随意，类似于随机算法

3).coordinato把组成员信息和订阅信息发送消费者

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 1]

*  protocol\_metadata: 序列化后的消费者的订阅信息
 *  generation\_id： 年代信息，类似于zookeeper的epoch，对于每一轮 rebalance，generation\_id都会递增。
    
    主要用来保护consumer group。隔离无效的oﬀset提交。也就是上一轮的consumer成员无法提交oﬀset到新的consumer group中。
 *  leader\_id： 消费组中的消费者，coordinator会选择一个作为leader，其实就是个member\_id
 *  members：consumer group中全部的消费者的订阅信息（只有leader能收到，然后根据策略进行分区分配）
 *  member\_metadata 对应消费者的订阅信息

> 注：因为每个消费者都可以设置自己的分区分配策略，而对于消费组，只能有一种大家都赞成的分配策略。所以leader会在当前JoinGroup阶段发起组内投票：
> 
> 1.  每个consumer都会把自己支持的分区分配策略发送到coordinator
> 2.  coordinator收集到所有消费者的分配策略，组成一个候选集
> 3.  每个消费者需要从候选集里找出一个自己支持的策略，并且为这个策略投票
> 4.  最终计算候选集中各个策略的选票数，票数最多的就是当前消费组的分配策略

### 3.3 Synchronizing Group State ###

完成分区分配之后，就进入了Synchronizing Group State阶段，主要逻辑是向GroupCoordinator发送 SyncGroupRequest请求，然后处理SyncGroupResponse响应

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 2]

简单来说，就是leader将消费者对应的partition分配方案同发送给coordinator，然后又coordinator将该方案设置到Response中发送给consumer group中的所有consumer。

> 两点注意：
> 
>  *  所有Consumer都会想coordinator发送syncgroup请求，但只有leader会发送分配方案
>  *  consumer group的分区分配方案是在客户端执行的！Kafka将这个权利下放给客户端主要是因为这样做可以有更好的灵活性

## 4.offset（细节） ##

oﬀset，他类似一个游标，表示当前消费的消息的位置。

每个topic可以划分多个分区（每个Topic至少有一个分区），同一topic下的不同分区包含的消息是不同的。每个消息在被添加到分区时，都会被分配一个 oﬀset（称之为偏移量），它是消息在此分区中的唯一编号，kafka通过oﬀset保证消息在分区内的顺序，oﬀset的顺序不跨分区，即kafka只保证在同一个分区内的消息是有序的；

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 3]

对于应用层消费者来说，每次消费一个消息并且提交后，会保存当前消费到的最近的一个oﬀset。那么oﬀset保存在哪里？

老版本的kafka中，offet是保存在zookeeper上。而现在的kafka中，提供了一个\_\_consumer\_oﬀsets 的一个topic，保存每个consumer group某一时刻提交的oﬀset信息。

*  consumer\_oﬀsets 默认有50个分区`(__cosumer_offsets-0, __cosumer_offsets-50)`  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 4]
 *  每个group具体分区计算公式：`Math.abs(groupId.hashCode()) % groupMetadataTopicPartitionCount`
    
     *  由于\_\_consumer\_oﬀsetst默认有50个分区，所以groupMetadataTopicPartitionCount默认=50
     *  若不同groupId计算出相同分区，怎么存？不影响，因为里面所有日志信息保存了其对应groupId
    
    ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 5][]

示例：group.id=KafkaConsumerDemo，所以 consumer\_oﬀsets中的分区 = Math.abs(“KafkaConsumerDemo”.hashCode()) % 50 = 35。意味着当前的 consumer\_group的位移信息保存在 consumer\_oﬀsets的第35个分区

另外，可以通过kafka-console-consumer.sh查看当前consumer group的offset具体提交信息。执行如下命令，可以查看当前consumer\_goup中的oﬀset位移提交的信息

sh kafka-console-consumer.sh --topic __consumer_offsets --partition 35 -bootstrap -server 
    localhost:9092 --formatter 'kafka.coordinator.group.GroupMetadataManager$OffsetsMessageFormatter'

[Kafka_Consumer]: https://blog.csdn.net/weixin_43935927/article/details/109697632
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/26224024b81f44f39ec21a270d729f9f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/705103c400664d66afb556fb75858563.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 2]: https://img-blog.csdnimg.cn/20201114233224318.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 3]: https://img-blog.csdnimg.cn/20201114233340756.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 4]: https://img-blog.csdnimg.cn/20201114233439683.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 5]: https://img-blog.csdnimg.cn/20201114233515209.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw==,size_16,color_FFFFFF,t_70#pic_center