RocketMQ-4.0-消息堆积

消息堆积问题

一般认为单条队列消息差值>=10w条的时候，就算堆积问题

什么情况下出现消息堆积？

1. 生产者生产消息太快了

   • 增加消费者数量，但是消费者数量<=队列数量

   • 适当地设置最大的消费线程数量

     /**
      * consumeThreadNumber是怎么计算的？
      * 线程池里面的最大核心线程数量（n是当前程序运行电脑的最大处理器数量）：
      *      IO密集型：2n
      *          读取数据、操作数据库等操作
      *      CPU密集型：n或者n+1
      *          用于计算、执行循环等操作
      */
     @Component
     @RocketMQMessageListener(
             topic = "backlogTopic",
             consumerGroup = "backlog-consumer-group",
             // 消息堆积的处理：适当地设置最大的消费线程数量（根据IO（2n）/CPU（n+1））
             consumeThreadNumber = 32
     )
     public class EBacklogListener1 implements RocketMQListener<String> {
         @Override
         public void onMessage(String message) {
             System.out.println("我是测试积压的第一个消费者：" + message);
         }
     }

     这里的consumeThreadNumber是根据当前机器的逻辑处理器来算出的，就是公式里的n

     

   • 动态扩容队列数量，从而增加消费者数量

     

     注意：扩容可以在消息没消费完的时候进行，但是缩容不行，缩容只能在消息消费完后进行设置，不然会导致消息丢失不会被消费

2. 消费者消费出现问题

   • 排查消费者程序的问题

跳过堆积

当消息出现堆积问题且不想处理的话可以对其进行跳过堆积处理（不推荐）