Kafka 位移主题
- 位移格式
- 创建位移
- 提交位移
- 删除位移
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 线程的状态
Compact 过期策略 ：
- 同个 Key 的两条消息 M1 和 M2，当 M1 发送时间早于 M2，那 M1 为过期消息
Compact 过程 :
- 扫描日志的所有消息，剔除那些过期的消息，把剩下的消息整理在一起
- 位移为 0、2 和 3 的消息的 Key 都是 K1，Compact 后，只会保存位移为 3 的消息
