台州seo网站推广,旅游网站信息门户建设方案,网页设计的设计思路怎么写,服务器怎么放网站吗Kafka中神秘的内部主题#xff08;Internal Topic#xff09;__consumer_offsets。 consumer_offsets在Kafka源码中有个更为正式的名字#xff0c;叫*位移主题*#xff0c;即Offsets Topic。为了方便今天的讨论#xff0c;我将统一使用位移主题来指代consumer_offsets。需…Kafka中神秘的内部主题Internal Topic__consumer_offsets。 consumer_offsets在Kafka源码中有个更为正式的名字叫*位移主题*即Offsets Topic。为了方便今天的讨论我将统一使用位移主题来指代consumer_offsets。需要注意的是它有两个下划线哦。 好了我们开始今天的内容吧。首先我们有必要探究一下位移主题被引入的背景及原因即位移主题的前世今生。 在上一期中我说过老版本Consumer的位移管理是依托于Apache ZooKeeper的它会自动或手动地将位移数据提交到ZooKeeper中保存。当Consumer重启后它能自动从ZooKeeper中读取位移数据从而在上次消费截止的地方继续消费。这种设计使得Kafka Broker不需要保存位移数据减少了Broker端需要持有的状态空间因而有利于实现高伸缩性。 但是ZooKeeper其实并不适用于这种高频的写操作因此Kafka社区自0.8.2.x版本开始就在酝酿修改这种设计并最终在新版本Consumer中正式推出了全新的位移管理机制自然也包括这个新的位移主题。 新版本Consumer的位移管理机制其实也很简单就是**将Consumer的位移数据作为一条条普通的Kafka消息提交到**consumer_offsets中。可以这么说**consumer_offsets的主要作用是保存Kafka消费者的位移信息。**它要求这个提交过程不仅要实现高持久性还要支持高频的写操作。显然Kafka的主题设计天然就满足这两个条件因此使用Kafka主题来保存位移这件事情实际上就是一个水到渠成的想法了。 这里我想再次强调一下和你创建的其他主题一样位移主题就是普通的Kafka主题。你可以手动地创建它、修改它甚至是删除它。只不过它同时也是一个内部主题大部分情况下你其实并不需要“搭理”它也不用花心思去管理它把它丢给Kafka就完事了。 虽说位移主题是一个普通的Kafka主题但它的消息格式却是Kafka自己定义的用户不能修改也就是说你不能随意地向这个主题写消息因为一旦你写入的消息不满足Kafka规定的格式那么Kafka内部无法成功解析就会造成Broker的崩溃。事实上Kafka Consumer有API帮你提交位移也就是向位移主题写消息。你千万不要自己写个Producer随意向该主题发送消息。 你可能会好奇这个主题存的到底是什么格式的消息呢所谓的消息格式你可以简单地理解为是一个KV对。Key和Value分别表示消息的键值和消息体在Kafka中它们就是字节数组而已。想象一下如果让你来设计这个主题你觉得消息格式应该长什么样子呢我先不说社区的设计方案我们自己先来设计一下。 首先从Key说起。一个Kafka集群中的Consumer数量会有很多既然这个主题保存的是Consumer的位移数据那么消息格式中必须要有字段来标识这个位移数据是哪个Consumer的。这种数据放在哪个字段比较合适呢显然放在Key中比较合适。 现在我们知道该主题消息的Key中应该保存标识Consumer的字段那么当前Kafka中什么字段能够标识Consumer呢还记得之前我们说Consumer Group时提到的Group ID吗没错就是这个字段它能够标识唯一的Consumer Group。 说到这里我再多说几句。除了Consumer GroupKafka还支持独立Consumer也称Standalone Consumer。它的运行机制与Consumer Group完全不同但是位移管理的机制却是相同的。因此即使是Standalone Consumer也有自己的Group ID来标识它自己所以也适用于这套消息格式。 Okay我们现在知道Key中保存了Group ID但是只保存Group ID就可以了吗别忘了Consumer提交位移是在分区层面上进行的即它提交的是某个或某些分区的位移那么很显然Key中还应该保存Consumer要提交位移的分区。 好了我们来总结一下我们的结论。位移主题的Key中应该保存3部分内容。如果你认同这样的结论那么恭喜你社区就是这么设计的 接下来我们再来看看消息体的设计。也许你会觉得消息体应该很简单保存一个位移值就可以了。实际上社区的方案要复杂得多比如消息体还保存了位移提交的一些其他元数据诸如时间戳和用户自定义的数据等。保存这些元数据是为了帮助Kafka执行各种各样后续的操作比如删除过期位移消息等。但总体来说我们还是可以简单地认为消息体就是保存了位移值。 当然了位移主题的消息格式可不是只有这一种。事实上它有3种消息格式。除了刚刚我们说的这种格式还有2种格式 用于保存Consumer Group信息的消息。用于删除Group过期位移甚至是删除Group的消息。 第1种格式非常神秘以至于你几乎无法在搜索引擎中搜到它的身影。不过你只需要记住它是用来注册Consumer Group的就可以了。 第2种格式相对更加有名一些。它有个专属的名字tombstone消息即墓碑消息也称delete mark。下次你在Google或百度中见到这些词不用感到惊讶它们指的是一个东西。这些消息只出现在源码中而不暴露给你。它的主要特点是它的消息体是null即空消息体。 那么何时会写入这类消息呢一旦某个Consumer Group下的所有Consumer实例都停止了而且它们的位移数据都已被删除时Kafka会向位移主题的对应分区写入tombstone消息表明要彻底删除这个Group的信息。 好了消息格式就说这么多下面我们来说说位移主题是怎么被创建的。通常来说当Kafka集群中的第一个Consumer程序启动时Kafka会自动创建位移主题。我们说过位移主题就是普通的Kafka主题那么它自然也有对应的分区数。但如果是Kafka自动创建的分区数是怎么设置的呢这就要看Broker端参数offsets.topic.num.partitions的取值了。它的默认值是50因此Kafka会自动创建一个50分区的位移主题。如果你曾经惊讶于Kafka日志路径下冒出很多__consumer_offsets-xxx这样的目录那么现在应该明白了吧这就是Kafka自动帮你创建的位移主题啊。 你可能会问除了分区数副本数或备份因子是怎么控制的呢答案也很简单这就是Broker端另一个参数offsets.topic.replication.factor要做的事情了。它的默认值是3。 总结一下如果位移主题是Kafka自动创建的那么该主题的分区数是50副本数是3。 当然你也可以选择手动创建位移主题具体方法就是在Kafka集群尚未启动任何Consumer之前使用Kafka API创建它。手动创建的好处在于你可以创建满足你实际场景需要的位移主题。比如很多人说50个分区对我来讲太多了我不想要这么多分区那么你可以自己创建它不用理会offsets.topic.num.partitions的值。 不过我给你的建议是还是让Kafka自动创建比较好。目前Kafka源码中有一些地方硬编码了50分区数因此如果你自行创建了一个不同于默认分区数的位移主题可能会碰到各种各样奇怪的问题。这是社区的一个Bug目前代码已经修复了但依然在审核中。 创建位移主题当然是为了用的那么什么地方会用到位移主题呢我们前面一直在说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过程。 图中位移为0、2和3的消息的Key都是K1。Compact之后分区只需要保存位移为3的消息因为它是最新发送的。 Kafka提供了专门的后台线程定期地巡检待Compact的主题看看是否存在满足条件的可删除数据。这个后台线程叫Log Cleaner。很多实际生产环境中都出现过位移主题无限膨胀占用过多磁盘空间的问题如果你的环境中也有这个问题我建议你去检查一下Log Cleaner线程的状态通常都是这个线程挂掉了导致的。 小结 总结一下今天我跟你分享了Kafka神秘的位移主题__consumer_offsets包括引入它的契机与原因、它的作用、消息格式、写入的时机以及管理策略等这对我们了解Kafka特别是Kafka Consumer的位移管理是大有帮助的。实际上将很多元数据以消息的方式存入Kafka内部主题的做法越来越流行。除了Consumer位移管理Kafka事务也是利用了这个方法当然那是另外的一个内部主题了。 社区的想法很简单既然Kafka天然实现了高持久性和高吞吐量那么任何有这两个需求的子服务自然也就不必求助于外部系统用Kafka自己实现就好了。