Kafka核心技术与实战——16 | 揭开神秘的“位移主题”面纱

Posted 2021-02-18 minimalist

• __consumer_offsets 在 Kafka 源码中有个更为正式的名字，叫位移主题，即 Offsets Topic。需要注意的是，它有两个下划线哦
  • 将 Consumer 的位移数据作为一条条普通的 Kafka 消息，提交到 __consumer_offsets 中
  • 可以这么说，__consumer_offsets 的主要作用是保存 Kafka 消费者的位移信息
• 和你创建的其他主题一样，位移主题就是普通的 Kafka 主题
  • 它的消息格式却是 Kafka 自己定义的
  • 用户不能修改，也就是说你不能随意地向这个主题写消息，因为一旦你写入的消息不满足 Kafka 规定的格式，那么 Kafka 内部无法成功解析，就会造成 Broker 的崩溃
  • 事实上，Kafka Consumer 有 API 帮你提交位移，也就是向位移主题写消息
  • 你千万不要自己写个 Producer 随意向该主题发送消息
• 所谓的消息格式，你可以简单地理解为是一个 KV 对
  • Key 和 Value 分别表示消息的键值和消息体，在 Kafka 中它们就是字节数组而已
  • 一个 Kafka 集群中的 Consumer 数量会有很多，既然这个主题保存的是 Consumer 的位移数据，那么消息格式中必须要有字段来标识这个位移数据是哪个 Consumer 的
  • 现在我们知道该主题消息的 Key 中应该保存标识 Consumer 的字段
  • 它能够标识唯一的 Consumer Group
• Consumer 提交位移是在分区层面上进行的，即它提交的是某个或某些分区的位移，那么很显然，Key 中还应该保存 Consumer 要提交位移的分区
  • 位移主题的 Key 中应该保存 3 部分内容：<Group ID，主题名，分区号 >
• 再来看看消息体的设计
  • 比如消息体还保存了位移提交的一些其他元数据，诸如时间戳和用户自定义的数据等
• 下面我们来说说位移主题是怎么被创建的
  • 当 Kafka 集群中的第一个 Consumer 程序启动时，Kafka 会自动创建位移主题
  • 如果位移主题是 Kafka 自动创建的，那么该主题的分区数是 50，副本数是 3
• 目前 Kafka Consumer 提交位移的方式有两种：自动提交位移和手动提交位移
  • Consumer 端有个参数叫 enable.auto.commit，如果值是 true，则 Consumer 在后台默默地为你定期提交位移，提交间隔由一个专属的参数 auto.commit.interval.ms来控制
  • 事实上，很多与 Kafka 集成的大数据框架都是禁用自动提交位移的，如 Spark、Flink 等。这就引出了另一种位移提交方式：手动提交位移，即设置 enable.auto.commit = false
  • Kafka Consumer API 为你提供了位移提交的方法，如 consumer.commitSync 等。当调用这些方法时，Kafka 会向位移主题写入相应的消息
• 如果你选择的是自动提交位移，那么就可能存在一个问题
  • 只要 Consumer 一直启动着，它就会无限期地向位移主题写入消息
  • 这就要求 Kafka 必须要有针对位移主题消息特点的消息删除策略，否则这种消息会越来越多，最终撑爆整个磁盘
• Kafka 是怎么删除位移主题中的过期消息的呢？
  • 答案就是 Compaction
  • 干脆采用 JVM 垃圾回收中的术语：整理
• Kafka 使用Compact 策略来删除位移主题中的过期消息，避免该主题无限期膨胀
  • 对于同一个 Key 的两条消息 M1 和 M2，如果 M1 的发送时间早于 M2，那么 M1 就是过期消息
  • Compact 的过程就是扫描日志的所有消息，剔除那些过期的消息，然后把剩下的消息整理在一起
  • 
• Kafka 提供了专门的后台线程定期地巡检待 Compact 的主题，看看是否存在满足条件的可删除数据
  • 这个后台线程叫 Log Cleaner。很多实际生产环境中都出现过位移主题无限膨胀占用过多磁盘空间的问题
  • 如果你的环境中也有这个问题，我建议你去检查一下 Log Cleaner 线程的状态，通常都是这个线程挂掉了导致的
• 小结
  • 总结一下，今天我跟你分享了 Kafka 神秘的位移主题 __consumer_offsets，包括引入它的契机与原因、它的作用、消息格式、写入的时机以及管理策略等，这对我们了解 Kafka 特别是 Kafka Consumer 的位移管理是大有帮助的
  • 社区的想法很简单：既然 Kafka 天然实现了高持久性和高吞吐量，那么任何有这两个需求的子服务自然也就不必求助于外部系统，用 Kafka 自己实现就好了