消息堆积对业务的影响

过量的消息堆积可能会引起内存或磁盘告警，从而造成所有connection阻塞，进而影响到其他队列的使用，导致整体服务质量的下降。

消息堆积产生的原因

1. 一般来说消息堆积是由于生产消息的速率远大于消费消息的速率所导致的。比如某个时间段消费端处理消息异常缓慢，发送一条消息只要3秒钟，而消费一条消息需要1分钟，每分钟发送20个消息，只有一个消息被消费端处理，这样队列中就会产生大量的消息堆积。
2. 消费者出现异常，生产者一直在发送消息，但是消费者不能消费，造成消息积压。
3. 消费者没有出现异常，但是消费者与队列间的订阅可能出现了异常，也会导致消息无法被消费从而造成堆积的情况。
4. 消费者正常，与队列间的订阅也正常，但是消费端的代码本身逻辑耗费时间长导致了消费能力降低，这时候就会出现1中的情况从而导致消息堆积。

解决消息堆积的办法

1. 生产速率较快，消费速率较慢 ：您可以通过以下方法解决。

• 增加消费者数量提高消费速率。
• 采用生产者确认的发送模式，并监控生产端消息生产速度和时长，当消息生产时长有明显增加时进行流控措施。

2. 消费者异常 ：建议排查消费者逻辑是不是有问题，优化程序。
3. 消费者与队列间的订阅异常 ：建议排查队列和消费者之间的订阅是否正常。
4. 消费端的代码本身逻辑耗费时间长 ：建议给消息设置过期时间，设置方法如下：

• 在生产消息时，设置消息过期时间。消息过期时间以expiration值体现。
  • 在properties中设置expiration值，单位为毫秒(ms)。

    AMQP.BasicProperties properties = new AMQP.BasicProperties().builder()
        .deliveryMode(2)
        .contentEncoding("UTF-8")
        .expiration("10000")
        .build();
    
    String message = "hello rabbitmq";
    channel.basicPublish(exchange, routingKey, properties, message.getBytes(StandardCharsets.UTF_8));
    

  • 在Web界面中设置expiration值，单位为毫秒(ms)。

登录Web界面，在“Exchanges”页签，单击Exchange名称，进入Exchange详情页。在“Publish message”区域，设置expiration值，如下图所示。

• 设置队列过期时间。队列过期时间以x-message-ttl值体现。从消息进入队列开始计算，超过了配置的队列过期时间，消息会自动被删除。

  • 在客户端代码中设置x-message-ttl值，单位为毫秒(ms)。

    Map<String, Object> arguments = new HashMap<String, Object>();
    arguments.put("x-message-ttl", 10000);
    channel.queueDeclare(queueName, true, false, false, arguments);
    

  • 在Web界面新建队列时，设置x-message-ttl值，单位为毫秒(ms)。

    登录Web界面，在“Exchanges”页签，新建队列时，设置x-message-ttl值，如下图所示。