Kafka 的内部主题 (Internal Topic) : __consumer_offsets (位移主题，Offsets Topic)

老 Consumer 会将位移消息提交到 ZK 中保存

• 当 Consumer 重启后，能自动从 ZK 中读取位移数据，继续消费上次位置
• Broker 不用保存位移数据，减少 Broker 开销
• 但 ZK 不适合大量写操作

新 Consumer 推出了位移管理机制 :

• 将 Consumer 的位移数据提交到 __consumer_offsets 中
• __consumer_offsets 作用 : 保存 Kafka 消费者的位移信息

位移格式

__consumer_offsets 的消息格式是 Kafka 自定义

• 不要随意向该主题写消息，可能会造成 Broker 挂
• Consumer API 会自动向位移主题写消息

位移主题的 3 种消息格式 :

• Key/Value 分别为消息键/消息体
• 保存 Consumer Group 信息的消息 : 用来注册 Consumer Group
• 删除 Group 过期位移 , 删除 Group 的消息

Key/Value 结构 :

• Key 的 3 部分：<Group ID, 主题名, 分区号>
• Value 有：时间戳 , 用户自定义的数据 , 位移值

删除 Group 消息 :

• tombstone 消息 (墓碑消息 , delete mark) ：特点 : 空消息体 , 消息体是 null
• 当某个 Group 下的所有 Consumer 都停止，且位移数据都已被删除 (彻底删除该 Group) ：Kafka 会向位移主题的对应分区写入 tombstone 消息

创建位移

位移主题自动创建 :

• 当 Kafka 的第一个 Consumer 启动时，Kafka 会自动创建位移主题
• 位移主题的分区数 : offsets.topic.num.partitions，默认值 : 50
• 副本数 : offsets.topic.replication.factor , 默认值 : 3

Kafka 日志路径下会有很多 __consumer_offsets-xxx 的目录

• Kafka 创建的位移主题

手动创建位移主题 ：不建议 (bug 代码有硬编码 50 )

提交位移

Consumer 提交位移方式：

• 自动提交位移
• 手动提交位移

自动提交位移 :

• Consumer 在后台定期提交位移
• 自动提交 ：enable.auto.commit=true
• 提交间隔 : auto.commit.interval.ms
• 优点 : 不用管位移提交，就能保证消息消费不会丢失
• 缺点 : 没法把控 Consumer 端的位移管理 ; 只要 Consumer 启动 , 就会不断向位移主题写入消息

与 Kafka 集成的框架都禁用手动提交位移

• enable.auto.commit = false
• Consumer 用 consumer.commitSync ，向位移主题写入相应的消息

自动提交位移的问题例子 :

• Consumer 消费到某个主题的最新一条消息 (位移 : 100)
• 之后该主题没有新消息产生，所以 Consumer 无消息可消费，则位移一直是 100
• 而自动提交位移，向位移主题中不断写位移 =100

删除位移

Compaction : Kafka 删除位移主题的过期消息

Kafka 用后台线程 (Log Cleaner) 定期检查 Compact 的主题，判断是否有可删除数据

• 当位移主题占用过多磁盘时，建议检查 Log Cleaner 线程的状态

查看位移主题的磁盘大小 ：

du -h /kafka/logs/__consumer_offsets-0/

Compact 过期策略 ：

• 同个 Key 的两条消息 M1 和 M2，当 M1 发送时间早于 M2，那 M1 为过期消息

Compact 过程 :

• 扫描日志的所有消息，剔除那些过期的消息，把剩下的消息整理在一起
• 位移为 0、2 和 3 的消息的 Key 都是 K1，Compact 后，只会保存位移为 3 的消息