【kafka原理】消费者偏移量__consumer_offsets_相关解析

刺骨的言语ヽ痛彻心扉 2022-12-18 15:49 122阅读 0赞

我们在kafka的log文件中发现了还有很多以 `__consumer_offsets_`的文件夹；总共50个;

由于Zookeeper并不适合大批量的频繁写入操作，新版Kafka已推荐将consumer的位移信息保存在Kafka内部的topic中，即`__consumer_offsets` topic，并且默认提供了`kafka_consumer_groups.sh`脚本供用户查看consumer信息。

`__consumer_offsets` 是 kafka 自行创建的，和普通的 topic 相同。它存在的目的之一就是保存 consumer 提交的位移。  
`__consumer_offsets` 的每条消息格式大致如图所示  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70_pic_center]  
可以想象成一个 KV 格式的消息，key 就是一个三元组：`group.id+topic+分区号`，而 value 就是 offset 的值。

考虑到一个 kafka 生成环境中可能有很多`consumer` 和 `consumer group`，如果这些 consumer 同时提交位移，则必将加重 \_\_consumer\_offsets 的写入负载，因此 kafka 默认为该 topic 创建了50个分区，并且对每个 `group.id`做哈希求模运算`Math.abs(groupID.hashCode()) % numPartitions`，从而将负载分散到不同的 \_\_consumer\_offsets 分区上。

一般情况下，当集群中第一次有消费者消费消息时会自动创建`__consumer_offsets`，它的副本因子受 `offsets.topic.replication.factor` 参数的约束，默认值为3（注意：该参数的使用限制在0.11.0.0版本发生变化），分区数可以通过 `offsets.topic.num.partitions` 参数设置，默认值为50。

### 1. 消费Topic消息 ###

打开一个session a，执行下面的消费者命令 ;指定了`消费组:szz1-group`; `topic:szz1-test-topic`

bin/kafka-console-consumer.sh --bootstrap-server  xxx1:9092,xxx2:9092,xxx3:9092 --group szz1-group --topic szz1-test-topic

### 2.产生消息 ###

打开一个新的session b,执行生产消息命令

bin/kafka-console-producer.sh --broker-list  xxx1:9092,xxx2:9092,xxx3:9092  --topic szz1-test-topic

发送几条消息  
![在这里插入图片描述][202010271124076.png_pic_center]

然后可以看到刚刚打开的 session a 消费了消息;  
![在这里插入图片描述][20201027112455861.png_pic_center]

### 3. 查看指定消费组的消费位置offset ###

bin/kafka-consumer-groups.sh --bootstrap-server xxx1:9092,xxx2:9092,xxx3:9092 --describe --group szz1-group

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70_pic_center 1]  
可以看到图中 展示了每个`partition` 对应的消费者id; 因为只开了一个消费者; 所以是这个消费者同时消费3个`partition`;  
`CURRENT-OFFSET: 当前消费组消费到的偏移量`  
`LOG-END-OFFSET: 日志最后的偏移量`  
`CURRENT-OFFSET` = `LOG-END-OFFSET` 说明当前消费组已经全部消费了;

那么我把 session a 关掉;现在没有消费者之后; 我再发送几条消息看看;  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70_pic_center 2]  
我发送了2条消息之后, `partition-0` `partition-1` 的`LOG-END-OFFSET: 日志最后的偏移量`分别增加了1; 但是`CURRENT-OFFSET: 当前消费组消费到的偏移量` 保持不变;因为没有被消费;

**重新打开一个消费组 继续消费**\*

重新打开session之后, 会发现控制台输出了刚刚发送的2条消息; 并且偏移量也更新了  
![在这里插入图片描述][20201027113806535.png_pic_center]

### 4. 从头开始消费 --from-beginning ###

如果我们用新的消费组去消费一个Topic,那么默认这个消费组的offset会是最新的; 也就是说历史的不会消费  
例如下面我们新开一个session c ;消费组设置为`szz1-group3`

bin/kafka-console-consumer.sh --bootstrap-server   xxx1:9092,xxx2:9092,xxx3:9092 --group szz1-group3    --topic szz1-test-topic

查看消费情况

bin/kafka-consumer-groups.sh --bootstrap-server  xxx1:9092,xxx2:9092,xxx3:9092  --describe --group szz1-group3

![在这里插入图片描述][20201027123506994.png_pic_center]  
可以看到`CURRENT-OFFSET` = `LOG-END-OFFSET` ;

如何让新的消费组/者 从头开始消费呢？ 加上参数 `--from-beginning`

### 5.如何确认 consume\_group 在哪个\_\_consumer\_offsets-? 中 ###

`Math.abs(groupID.hashCode()) % numPartitions`

### 6. 查找\_\_consumer\_offsets 分区数中的消费组偏移量offset ###

上面的 **3. 查看指定消费组的消费位置offset** 中,我们知道如何查看指定的topic消费组的偏移量;  
那还有一种方式也可以查询

先通过 `consume_group` 确定分区数; 例如 `"szz1-group".hashCode()%50=32`; 那我们就知道 `szz-group`消费组的偏移量信息存放在 `__consumer_offsets_32`中;  
通过命令

bin/kafka-simple-consumer-shell.sh --topic __consumer_offsets --partition 32 --broker-list xxx1:9092,xxx2:9092,xxx3:9092 --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter"

![在这里插入图片描述][20201027135823471.png_pic_center]  
**前面的 是key 后面的是value；key由 消费组+Topic+分区数 确定; 后面的value就包含了 消费组的偏移量信息等等**

然后接着我们发送几个消息,并且进行消费; 上面的控制台会自动更新为新的offset;

### 7 查询topic的分布情况 ###

bin/kafka-topics.sh --describe --zookeeper xxx:2181 --topic TOPIC名称

--------------------

> **日常运维**、**问题排查**=> [滴滴开源LogiKM一站式Kafka监控与管控平台][LogiKM_Kafka]

--------------------

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

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70_pic_center]: /images/20221122/0cb151bb80264b4d9110c7c107442c4c.png
[202010271124076.png_pic_center]: /images/20221122/19696797c53640be9bf2c01f95d71582.png
[20201027112455861.png_pic_center]: /images/20221122/9a26ef84e0b245ab8f58cf0c1e7d8ab2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221122/07305fd1cc984b4e8b98aac91c89bde8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221122/e7a46b9edd2b4511ad0b3e34ae963d3a.png
[20201027113806535.png_pic_center]: /images/20221122/5c285f4356314602a683c1f3d47f610c.png
[20201027123506994.png_pic_center]: /images/20221122/f7071e795f1947f8a953c1680a24aeff.png
[20201027135823471.png_pic_center]: /images/20221122/7c174a67106b4612b8757d38a2a6b7fc.png
[LogiKM_Kafka]: https://github.com/didi/LogiKM
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221122/156165a39d3d402386970af0a0414372.png