kafka offset __consumer_offsets_ 原理/Compact 整理

「爱情、让人受尽委屈。」 2021-09-25 03:52 302阅读 1赞

我们在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

*  1

发送几条消息  
![在这里插入图片描述][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

*  1

![在这里插入图片描述][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

*  1

查看消费情况

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

*  1

![在这里插入图片描述][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名称

[https://blog.csdn.net/u010634066/article/details/109306637][https_blog.csdn.net_u010634066_article_details_109306637]

\--------------

**Kafka 集群中的第一个 Consumer 程序启动时，Kafka 会自动创建位移主题**。我们说过，位移主题就是普通的 Kafka 主题，那么它自然也有对应的分区数。但如果是 Kafka 自动创建的，分区数是怎么设置的呢？这就要看 Broker 端参数 offsets.topic.num.partitions 的取值了。它的默认值是 50，因此 Kafka 会自动创建一个 50 分区的位移主题。

总结一下，**如果位移主题是 Kafka 自动创建的，那么该主题的分区数是 50，副本数是 3**。

创建位移主题当然是为了用的，那么什么地方会用到位移主题呢？我们前面一直在说 Kafka Consumer 提交位移时会写入该主题，那 Consumer 是怎么提交位移的呢？目前 Kafka Consumer 提交位移的方式有两种：**自动提交位移和手动提交位移。**

Consumer 端有个参数叫 enable.auto.commit，如果值是 true，则 Consumer 在后台默默地为你定期提交位移，提交间隔由一个专属的参数 auto.commit.interval.ms 来控制。自动提交位移有一个显著的优点，就是省事，你不用操心位移提交的事情，就能保证消息消费不会丢失。但这一点同时也是缺点。因为它太省事了，以至于丧失了很大的灵活性和可控性，你完全没法把控 Consumer 端的位移管理。

事实上，很多与 Kafka 集成的大数据框架都是禁用自动提交位移的，如 Spark、Flink 等。这就引出了另一种位移提交方式：**手动提交位移**，即设置 enable.auto.commit = false。一旦设置了 false，作为 Consumer 应用开发的你就要承担起位移提交的责任。Kafka Consumer API 为你提供了位移提交的方法，如 consumer.commitSync 等。当调用这些方法时，Kafka 会向位移主题写入相应的消息。

如果你选择的是自动提交位移，那么就可能存在一个问题：只要 Consumer 一直启动着，它就会无限期地向位移主题写入消息。

我们来举个极端一点的例子。假设 Consumer 当前消费到了某个主题的最新一条消息，位移是 100，之后该主题没有任何新消息产生，故 Consumer 无消息可消费了，所以位移永远保持在 100。由于是自动提交位移，位移主题中会不停地写入位移 =100 的消息。显然 Kafka 只需要保留这类消息中的最新一条就可以了，之前的消息都是可以删除的。这就要求 Kafka 必须要有针对位移主题消息特点的消息删除策略，否则这种消息会越来越多，最终撑爆整个磁盘。

**Kafka 是怎么删除位移主题中的过期消息的呢？答案就是 Compaction。国内很多文献都将其翻译成压缩，我个人是有一点保留意见的。在英语中，压缩的专有术语是 Compression，它的原理和 Compaction 很不相同，我更倾向于翻译成压实，或干脆采用 JVM 垃圾回收中的术语：整理。**

不管怎么翻译，Kafka 使用**Compact 策略**来删除位移主题中的过期消息，避免该主题无限期膨胀。那么应该如何定义 Compact 策略中的过期呢？对于同一个 Key 的两条消息 M1 和 M2，如果 M1 的发送时间早于 M2，那么 M1 就是过期消息。Compact 的过程就是扫描日志的所有消息，剔除那些过期的消息，然后把剩下的消息整理在一起。我在这里贴一张来自官网的图片，来说明 Compact 过程。

![bb05c8c5fa90676a9456c775287dc358.png][]

图中位移为 0、2 和 3 的消息的 Key 都是 K1。Compact 之后，分区只需要保存位移为 3 的消息，因为它是最新发送的。

**Kafka 提供了专门的后台线程定期地巡检待 Compact 的主题，看看是否存在满足条件的可删除数据**。**这个后台线程叫 Log Cleaner。很多实际生产环境中都出现过位移主题无限膨胀占用过多磁盘空间的问题，如果你的环境中也有这个问题，我建议你去检查一下 Log Cleaner 线程的状态，通常都是这个线程挂掉了导致的。**

参考：[https://blog.csdn.net/qq\_41049126/article/details/111311816?utm\_medium=distribute.pc\_relevant.none-task-blog-baidujs\_title-2&spm=1001.2101.3001.4242][https_blog.csdn.net_qq_41049126_article_details_111311816_utm_medium_distribute.pc_relevant.none-task-blog-baidujs_title-2_spm_1001.2101.3001.4242]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70_pic_center]: /images/20210923/98b667c8559a4e26b43c1b37e112cccf.png
[202010271124076.png_pic_center]: /images/20210923/6b281317f2c545f8a169ad873323b6e7.png
[20201027112455861.png_pic_center]: /images/20210923/3c92d830a52a4eb282148e216b80a23f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20210923/16fa84f160a64576bb9634d33f3e4eb5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20210923/ea2d1d5d2bb3455eb4f02b3c5d0637ce.png
[20201027113806535.png_pic_center]: /images/20210923/70b92e17e34845eab596fb35f13b9cab.png
[20201027123506994.png_pic_center]: /images/20210923/35ace357b7634715b05176df729ba369.png
[20201027135823471.png_pic_center]: /images/20210923/c3f846b8ec7b465dbcb03cd726ca33ea.png
[https_blog.csdn.net_u010634066_article_details_109306637]: https://blog.csdn.net/u010634066/article/details/109306637
[bb05c8c5fa90676a9456c775287dc358.png]: /images/20210923/3b40b29081b549c9bf1389c053ff1e52.png
[https_blog.csdn.net_qq_41049126_article_details_111311816_utm_medium_distribute.pc_relevant.none-task-blog-baidujs_title-2_spm_1001.2101.3001.4242]: https://blog.csdn.net/qq_41049126/article/details/111311816?utm_medium=distribute.pc_relevant.none-task-blog-baidujs_title-2&spm=1001.2101.3001.4242