RocketMQ 消费端限流机制

it2023-01-19  69

RocketMQ 消息消费端会从 3 个维度进行限流:

消息消费端队列中积压的消息超过 1000 条消息处理队列中尽管积压没有超过 1000 条,但最大偏移量与最小偏移量的差值超过 2000消息处理队列中积压的消息总大小超过 100M

为了方便理解上述三条规则的设计理念,我们首先来看一下消费端的数据结构,如下图所示:

PullMessageService 线程会按照队列向 Broker 拉取一批消息,然后会存入到 ProcessQueue 队列中,即所谓的处理队列,然后再提交到消费端线程池中进行消息消费,消息消费完成后会将对应的消息从 ProcessQueue 中移除,然后向 Broker 端提交消费进度,提交的消费偏移量为 ProceeQueue 中的最小偏移量。

规则一:消息消费端队列中积压的消息超过 1000 条值的就是 ProcessQueue 中存在的消息条数超过指定值,默认为 1000 条,就触发限流,限流的具体做法就是暂停向 Broker 拉取该队列中的消息,但并不会阻止其他队列的消息拉取。例如如果 q0 中积压的消息超过 1000 条,但 q1 中积压的消息不足 1000,那 q1 队列中的消息会继续消费。其目的就是担心积压的消息太多,如果再继续拉取,会造成内存溢出。

规则二:消息在 ProcessQueue 中实际上维护的是一个 TreeMap,key 为消息的偏移量、vlaue 为消息对象,由于 TreeMap 本身是排序的,故很容易得出最大偏移量与最小偏移量的差值,即有可能存在处理队列中其实就只有 3 条消息,但偏移量确超过了 2000,例如如下图所示:

出现这种情况也是非常有可能的,其主要原因就是消费偏移量为 100 的这个线程由于某种情况卡主了(“阻塞”了),其他消息却能正常消费,这种情况虽然不会造成内存溢出,但大概率会造成大量消息重复消费,究其原因与消息消费进度的提交机制有关,在 RocketMQ 中,例如消息偏移量为 2001 的消息消费成功后,向服务端汇报消费进度时并不是报告 2001,而是取处理队列中最小偏移量 100,这样虽然消息一直在处理,但消息消费进度始终无法向前推进,试想一下如果此时最大的消息偏移量为 1000,项目组发现出现了消息积压,然后重启消费端,那消息就会从 100 开始重新消费,会造成大量消息重复消费,RocketMQ 为了避免出现大量消息重复消费,故对此种情况会对其进行限制,超过 2000 就不再拉取消息了。

规则三:消息处理队列中积压的消息总大小超过 100M。

这个就更加直接了,不仅从消息数量考虑,再结合从消息体大小考虑,处理队列中消息总大小超过 100M 进行限流,这个显而易见就是为了避免内存溢出。

在了解了 RocketMQ 消息限流规则后,会在 rocketmq_client.log 中输出相关的限流日志,具体搜索“so do flow control”,详细如下图所示:

最新回复(0)