视频网站开发需要什么插件,网站开发和推广的不同,东莞市网络优化推广平台,企业官网的作用前言消息堆积是消息中间件的一大特色#xff0c;消息中间件的流量削峰、冗余存储等功能正是得益于消息中间件的消息堆积能力。然而消息堆积其实是一把亦正亦邪的双刃剑#xff0c;如果应用场合不恰当反而会对上下游的业务造成不必要的麻烦#xff0c;比如消息堆积势必会影响…前言消息堆积是消息中间件的一大特色消息中间件的流量削峰、冗余存储等功能正是得益于消息中间件的消息堆积能力。然而消息堆积其实是一把亦正亦邪的双刃剑如果应用场合不恰当反而会对上下游的业务造成不必要的麻烦比如消息堆积势必会影响上下游整个调用链的时效性有些中间件如RabbitMQ在发生消息堆积时在某些情况下还会影响自身的性能。对于Kafka而言虽然消息堆积不会对其自身性能带来多大的困扰但难免不会影响上下游的业务堆积过多有可能会造成磁盘爆满或者触发日志清除策略而造成消息丢失的情况。如何利用好消息堆积这把双刃剑监控是最为关键的一步。正文消息堆积是消费滞后(Lag)的一种表现形式消息中间件服务端中所留存的消息与消费掉的消息之间的差值即为消息堆积量也称之为消费滞后(Lag)量。对于Kafka而言消息被发送至Topic中而Topic又分成了多个分区(Partition)每一个Partition都有一个预写式的日志文件虽然Partition可以继续细分为若干个段文件(Segment)但是对于上层应用来说可以将Partition看成最小的存储单元(一个由多个Segment文件拼接的“巨型文件”)。每个Partition都由一系列有序的、不可变的消息组成这些消息被连续的追加到Partition中。我们来看下图其就是Partition的一个真实写照上图中有四个概念LogStartOffset表示一个Partition的起始位移初始为0虽然消息的增加以及日志清除策略的影响这个值会阶段性的增大。ConsumerOffset消费位移表示Partition的某个消费者消费到的位移位置。HighWatermark简称HW代表消费端所能“观察”到的Partition的最高日志位移HW大于等于ConsumerOffset的值。LogEndOffset简称LEO,代表Partition的最高日志位移其值对消费者不可见。比如在ISR(In-Sync-Replicas)副本数等于3的情况下(如下图所示)消息发送到LeaderA之后会更新LEO的值Follower B和Follower C也会实时拉取LeaderA中的消息来更新自己HW就表示A、B、C三者同时达到的日志位移也就是A、B、C三者中LEO最小的那个值。由于B、C拉取A消息之间延时问题所以HW必然不会一直与Leader的LEO相等即LEOHW。要计算Kafka中某个消费者的滞后量很简单首先看看其消费了几个Topic然后针对每个Topic来计算其中每个Partition的Lag每个Partition的Lag计算就显得非常的简单了参考下图由图可知消费LagHW - ConsumerOffset。对于这里大家有可能有个误区就是认为Lag应该是LEO与ConsumerOffset之间的差值笔者在这之前也犯过这样的错误认知详细可以参考《如何使用JMX监控Kafka》。LEO是对消费者不可见的既然不可见何来消费滞后一说。那么这里就引入了一个新的问题HW和ConsumerOffset的值如何获取呢首先来说说ConsumerOffsetKafka中有两处可以存储一个是Zookeeper而另一个是”__consumer_offsets这个内部topic中前者是0.8.x版本中的使用方式但是随着版本的迭代更新现在越来越趋向于后者。就拿1.0.0版本来说虽然默认是存储在”__consumer_offsets”中但是保不齐用于就将其存储在了Zookeeper中了。这个问题倒也不难解决针对两种方式都去拉取然后哪个有值的取哪个。不过这里还有一个问题对于消费位移来说其一般不会实时的更新而更多的是定时更新这样可以提高整体的性能。那么这个定时的时间间隔就是ConsumerOffset的误差区间之一。再来说说HW其也是Kafka中Partition的一个状态。有可能你会察觉到在Kafka的JMX中可以看到“kafka.log:typeLog,nameLogEndOffset,topic[topic_name],partition[partition_num]”这样一个属性但是这个值不是LEO而是HW。那么怎样正确的计算消费的Lag呢对Kafka熟悉的同学可能会想到Kafka中自带的kafka-consumer_groups.sh脚本中就有Lag的信息示例如下1[rootnode2 kafka_2.12-1.0.0]# bin/kafka-consumer-groups.sh --describe --bootstrap-server localhost:9092 --group CONSUMER_GROUP_ID2TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID3topic-test1 0 1648 1648 0 CLIENT_ID-e2d41f8d-dbd2-4f0e-9239-efacb55c6261 /192.168.92.1 CLIENT_ID4topic-test1 1 1648 1648 0 CLIENT_ID-e2d41f8d-dbd2-4f0e-9239-efacb55c6261 /192.168.92.1 CLIENT_ID5topic-test1 2 1648 1648 0 CLIENT_ID-e2d41f8d-dbd2-4f0e-9239-efacb55c6261 /192.168.92.1 CLIENT_ID6topic-test1 3 1648 1648 0 CLIENT_ID-e2d41f8d-dbd2-4f0e-9239-efacb55c6261 /192.168.92.1 CLIENT_ID71我们深究一下kafka-consumer_groups.sh脚本发现只有一句代码1exec $(dirname $0)/kafka-run-class.sh kafka.admin.ConsumerGroupCommand $其含义就是执行kafka.admin.ConsumerGroupCommand而已。进一步深究在ConsumerGroupCommand内部抓住了2句关键代码1val consumerGroupService new KafkaConsumerGroupService(opts)2val (state, assignments) consumerGroupService.describeGroup()代码详解consumerGroupService的类型是ConsumerGroupServicesealed trait类型)而KafkaConsumerGroupService只是ConsumerGroupService的一种实现还有一种实现是ZkConsumerGroupService分别对应新版的消费方式(消费位移存储在__consumer_offsets中)和旧版的消费方式(消费位移存储在zk中)详细计算步骤参考下一段落的内容。opt参数是指“ –describe –bootstrap-server localhost:9092 –group CONSUMER_GROUP_ID”等参数。第2句代码是调用describeGroup()方法来获取具体的信息即二元组中的assignments这个assignments中保存了上面打印信息中的所有内容。Scala小知识在Scala中trait(特征)相当于Java的接口实际上它比接口更大强大。与Java中的接口不同的是它还可以定义属性和方法的实现(JDK8起的接口默认方法)。一般情况下Scala中的类只能继承单一父类但是如果是trait的话就可以继承多个从结果来看是实现了多重继承。被sealed声明的trait仅能被同一文件的类继承。ZkConsumerGroupService中计算消费lag的步骤如下通过zk获取一些基本信息对应上面打印信息中的TOPIC、PARTITION、CONSUMER-ID等不过不会有HOST和CLIENT-ID。1、通过OffsetFetchRequest请求获取消费位移(offset)如果获取失败则在通过zk获取。2、通过OffsetReuqest请求获取分区的LogEndOffset(简称为LEO可见的LEO)。3、计算LogEndOffset与消费位移的差值来获取lag。4、KafkaConsumerGroupService中计算消费lag的步骤如下通过DescibeGroupsRequest请求获取一些基本信息不仅包括TOPIC、PARTITION、CONSUMER-ID还有HOST和CLIENT-ID。其实还有通过1、FindCoordinatorRequest请求来获取coordinator信息如果不了解coordinator在这里也没影响。2、通过OffsetFetchRequest请求获取消费位移。3、通过OffsetReuqest请求获取分区的LogEndOffset(简称为LEO)。4、计算LogEndOffset与消费位移的差值来获取lag。可以看到KafkaConsumerGroupService与ZkConsumerGroupService的计算Lag的方式都差不多但是KafkaConsumerGroupService能获取更多消费详情并且ZkConsumerGroupService也被标注为Deprecated的了后面内容都针对KafkaConsumerGroupService来做说明。既然Kafka已经为我们提供了线程的方法来获取Lag那么我们有何必再重复造轮子这里笔者写了一个调用的KafkaConsumerGroupService的示例(KafkaConsumerGroupService是使用Scala语言编写的在Java的程序里使用类似scala.collection.Seq这样的全名称以防止混淆) 1String[] agrs {--describe