RocketMQ消息堆积常见场景与处理方案

一、前言

在我的日常开发工作中，有碰到过一些使用RocketMQ消息堆积的场景，消息堆积是某个生产者发送了消息到topic下的队列中，但是消费者未及时消费，导致消费者组的消息堆积的情况，本文将分析常见的消息堆积的场景以及实际的处理方案。

二、消息堆积出现的场景

1、消费者注册未成功

这种情况发生在新上线消费者组，我们有遇到过消费组下的消费者实例设置的instanceName都一致的情况下，这个时候消费者组下某些消费者注册不成功，从而会发生消息堆积的情况。

2、消息生产过快，而消费者消费过慢

这种情况是我们的业务消费逻辑消费慢的场景，比如消费用户访问app首页消息，这种消息生产速度是非常快的，如果我们的消费逻辑非常慢的话，就出现消息堆积的情况。

三、消息堆积解决方案

• 如果出现了消息堆积的情况，我们需要怎么处理呢。

上面第一种情况是属于MQ使用姿势问题，消费者都没有注册上去，这种情况需要及时调整代码，将消费者注册好，对于堆积的消息进行抛弃或者加快消费。

• 如果是第二种情况，这是非使用姿势问题，消费者消费能力的问题。我们可以怎么处理呢？

手段一、提高消费并行度

1、同一个ConsumerGroup,通过增加 Consumer 实例数量来提高并行度（需要注意的是超过订阅队列数的 Consumer 实例无效）。可以通过加机器，或者在已有机器启动多个进程的方式。

2、提高单个 Consumer 的消费并行线程，通过修改参数 consumeThreadMin、consumeThreadMax实现。RocketMQ默认消费线程是20个。我们可以将他配置到分布式配置中心，比如apollo,实现动态的调整。

手段二、批量消费方式

某些业务流程如果支持批量方式消费，则可以很大程度上提高消费吞吐量，例如订单扣款类应用，一次处理一个订单耗时 1 s，一次处理 10 个订单可能也只耗时 2 s，这样即可大幅度提高消费的吞吐量。

通过设置 consumer的 consumeMessageBatchMaxSize 参数，默认是 1，即一次只消费一条消息，例如设置为 N，那么每次消费的消息数小于等于 N。

手段三、跳过非重要消息，追赶上生产者进度

发生消息堆积时，如果消费速度一直追不上发送速度，如果业务对数据要求不高的话，可以选择丢弃不重要的消息。例如，当某个队列的消息数堆积到100000条以上，则尝试丢弃部分或全部消息，这样就可以快速追上发送消息的速度。示例代码如下：

public ConsumeConcurrentlyStatus consumeMessage(
            List<MessageExt> msgs,
            ConsumeConcurrentlyContext context) {
        long offset = msgs.get(0).getQueueOffset();
        String maxOffset =
                msgs.get(0).getProperty(Message.PROPERTY_MAX_OFFSET);
        long diff = Long.parseLong(maxOffset) - offset;
        if (diff > 100000) {
            // TODO 消息堆积情况的特殊处理
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        // TODO 正常消费过程
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }   

手段四、优化消费代码业务逻辑

举例如下，某条消息的消费过程如下：

这条消息的消费过程中有4次与 DB的 交互，如果按照每次 5ms 计算，那么总共耗时 20ms，假设业务计算耗时 5ms，那么总过耗时 25ms，所以如果能把 4 次 DB 交互优化为 2 次，那么总耗时就可以优化到 15ms，即总体性能提高了 40%。所以应用如果对时延敏感的话，可以把DB部署在SSD硬盘，相比于SCSI磁盘，前者的RT会小很多。

四、总结

消息堆积通常由于使用姿势问题或者消费者消费速率过慢导致，其中消费者过慢的解决方案也有很多种，我们的解决实际问题期间需要根据实际的使用场景选择解决最快并且消耗更少资源的方案。

你有没有其他的方案呢？欢迎在评论区和JY分享。