• 0
• 0

记一次 RabbitMQ 消费者莫名消失问题的排查

技术分享 1年前 (2024-09-24) 0 999+

关注

开心一刻

今天好哥们找我借钱
哥们：兄弟，我最近手头紧，能不能借我点...
我：我手头也不宽裕，要不你试试银行贷款或者花呗？
哥们：不行，那个借了要还的
我：...

问题回顾

某天下午，生产监控告警：消息积压，队列 xxx 消息数超过 100；我第一时间想到的是应用服务是不是停了，但应用服务存活监控又没有告警，但我还是找值班运维同事帮忙确认了下，确认结果是服务的 6 个节点都是存活的，然后我又让运维确认了下队列的消费者情况，结果发现消费者列表中只有 2 个节点的消费者，其他 4 个节点的消费者不见了，所以消息消费不过来，导致了消息积压！

所以问题来了

那 4 个节点的注册的消费者为何消失？

但当务之急是解决消息积压的问题，所以让运维重启那 4 个节点的服务，消费者重新注册上，消息得以快速消费，消息积压告警得以恢复

生产问题虽暂时得以解决，但未找到根因，还是存在复发风险；下面就请大家跟随我的脚本，来看看我是如何排查的

问题排查

直接查 ERROR 级别的日志，很容易就能就找到了关键日志

Consumer thread error, thread abort.

以及异常堆栈

java.lang.OutOfMemoryError: Requested array size exceeds VM limit 	at java.lang.StringCoding$StringEncoder.encode(StringCoding.java:300) 	at java.lang.StringCoding.encode(StringCoding.java:344) 	at java.lang.String.getBytes(String.java:918)     ... 

Consumer thread error, thread abort 大家能看懂吧，就是字面意思

消费者线程错误，线程中止

消费者线程就是我们前面提到的队列消费者，一个队列消费者就是一个消费者线程，消费者线程中止那就意味着队列消费者中止，也就对应文章标题中的 消费者消失；是不是离真相越来越近了？

OutOfMemoryError 是不是很熟悉，内存溢出嘛

OutOfMemoryError 表示 Java 虚拟机在堆内存中没有足够的空间来分配对象

问你们一个问题：OOM 一定会导致 JVM 退出吗，这个问题你们先思考，后面答案会揭晓

回到正题，从关键日志以及异常堆栈，我们是不是可以得出以下推测

OOM 会导致消费者线程中止

有了推测，那就去验证呗；我先给大家模拟下案例，基于 SpringBoot，pom.xml 关键依赖

<dependencies>     <dependency>         <groupId>org.springframework.boot</groupId>         <artifactId>spring-boot-starter-amqp</artifactId>     </dependency>      <dependency>         <groupId>org.projectlombok</groupId>         <artifactId>lombok</artifactId>     </dependency> </dependencies> 

配置文件 application.yml

server:   port: 8088 spring:   rabbitmq:     host: 192.168.2.118     port: 5672     username: admin     password: admin     virtual-host: /     listener:       simple:         acknowledge-mode: manual #设置确认模式手工确认         concurrency: 3 #消费者个数，线程个数         prefetch: 1 

RabbitMQ 配置 TaskRabbitConfig.java

package com.qsl.rabbit.config;  import com.qsl.rabbit.constant.Constant; import com.qsl.rabbit.listener.TaskMessageListener; import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.DirectExchange; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration;  /**  * @author: 青石路  */ @Configuration public class TaskRabbitConfig {      @Value("${spring.rabbitmq.listener.simple.concurrency:3}")     private int concurrency;     @Value("${spring.rabbitmq.listener.simple.prefetch:1}")     private int prefetch;      @Bean     public DirectExchange taskExchange() {         return new DirectExchange(Constant.TASK_EXCHANGE, true, false);     }      @Bean     public Queue taskQueue() {         return new Queue(Constant.TASK_QUEUE, true, false, false);     }      @Bean     public Binding bindingTaskQueue() {         return BindingBuilder.bind(taskQueue()).to(taskExchange()).with(Constant.TASK_QUEUE);     }      @Bean     public SimpleMessageListenerContainer taskMessageListenerContainer(ConnectionFactory connectionFactory) {          SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();         container.setConnectionFactory(connectionFactory);         //设置确认模式手工确认         container.setAcknowledgeMode(AcknowledgeMode.MANUAL);         container.setQueueNames(Constant.TASK_QUEUE);         //消费者个数，线程个数         container.setConcurrentConsumers(concurrency);         //设置预处理个数         container.setPrefetchCount(prefetch);          container.setMessageListener(new TaskMessageListener());         return container;     } } 

消息监听器 TaskMessageListener.java

/**  * @author: 青石路  */ @Slf4j public class TaskMessageListener implements ChannelAwareMessageListener {      @Override     public void onMessage(Message message, Channel channel) {         String content = new String(message.getBody(), StandardCharsets.UTF_8);         log.info("消费者接收到消息：{}", content);         handleTask(content);         try {             // 手动ACK             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);         } catch (IOException e) {             log.error("消息确认失败，异常：", e);         }     }      private void handleTask(String message) {         try {             // 业务处理             log.info("处理任务：{}", message);             log.info("任务处理完成");         } catch (Exception e) {             log.error("处理任务失败，异常：", e);         }     } } 

业务处理的时候进行 Exception 捕获，并且手动确认消息，我相信你们平时都是这么用的，这难道有什么问题？我调整下 handleTask 方法

/**  * 业务处理  * @param message 消息内容  * @author: 青石路  */ private void handleTask(String message) {     try {         // 业务处理         log.info("处理任务：{}", message);         int i = 3 / (message.length() % 10);         if (i == 1) {             throw new OutOfMemoryError("模拟内存溢出");         }         log.info("任务处理结果：{}", i);     } catch (Exception e) {         log.error("处理任务失败，异常：", e);     } } 

启动服务后，队列消费者情况如下

发送消息 a，日志输出如下

2024-09-22 20:15:55|taskMessageListenerContainer-2|com.qsl.rabbit.listener.TaskMessageListener|INFO|20|消费者接收到消息：a 2024-09-22 20:15:55|taskMessageListenerContainer-2|com.qsl.rabbit.listener.TaskMessageListener|INFO|37|处理任务：a 2024-09-22 20:15:55|taskMessageListenerContainer-2|com.qsl.rabbit.listener.TaskMessageListener|INFO|42|任务处理结果：3 

相当于业务正常处理；我们再发送消息 abcdefghij，日志输出如下

2024-09-22 20:17:45|taskMessageListenerContainer-3|com.qsl.rabbit.listener.TaskMessageListener|INFO|20|消费者接收到消息：abcdefghij 2024-09-22 20:17:45|taskMessageListenerContainer-3|com.qsl.rabbit.listener.TaskMessageListener|INFO|37|处理任务：abcdefghij 2024-09-22 20:17:45|taskMessageListenerContainer-3|com.qsl.rabbit.listener.TaskMessageListener|ERROR|44|处理任务失败，异常： java.lang.ArithmeticException: / by zero 	at com.qsl.rabbit.listener.TaskMessageListener.handleTask(TaskMessageListener.java:38) 	at com.qsl.rabbit.listener.TaskMessageListener.onMessage(TaskMessageListener.java:21) 	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1591) 	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1510) 	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1498) 	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1489) 	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1433) 	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:975) 	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:921) 	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:83) 	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1296) 	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1202) 	at java.lang.Thread.run(Thread.java:748) 

被除数为 0，出现 ArithmeticException

相当于业务处理出现了 Exception，而我们进行了 catch，所以日志打印也符合我们的代码逻辑，也不会对消费者线程造成影响，队列消费者还是最初的那 3 个

我们发送消息 ab，日志输出如下

2024-09-22 20:36:31|taskMessageListenerContainer-1|com.qsl.rabbit.listener.TaskMessageListener|INFO|20|消费者接收到消息：ab 2024-09-22 20:36:31|taskMessageListenerContainer-1|com.qsl.rabbit.listener.TaskMessageListener|INFO|37|处理任务：ab 2024-09-22 20:36:31|taskMessageListenerContainer-1|org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer|ERROR|1268|Consumer thread error, thread abort. java.lang.OutOfMemoryError: 模拟内存溢出 	at com.qsl.rabbit.listener.TaskMessageListener.handleTask(TaskMessageListener.java:40) 	at com.qsl.rabbit.listener.TaskMessageListener.onMessage(TaskMessageListener.java:21) 	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1591) 	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1510) 	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1498) 	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1489) 	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1433) 	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:975) 	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:921) 	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:83) 	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1296) 	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1202) 	at java.lang.Thread.run(Thread.java:748) 2024-09-22 20:36:31|taskMessageListenerContainer-2|com.qsl.rabbit.listener.TaskMessageListener|INFO|20|消费者接收到消息：ab 2024-09-22 20:36:31|taskMessageListenerContainer-2|com.qsl.rabbit.listener.TaskMessageListener|INFO|37|处理任务：ab 2024-09-22 20:36:31|taskMessageListenerContainer-2|org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer|ERROR|1268|Consumer thread error, thread abort. java.lang.OutOfMemoryError: 模拟内存溢出 	at com.qsl.rabbit.listener.TaskMessageListener.handleTask(TaskMessageListener.java:40) 	at com.qsl.rabbit.listener.TaskMessageListener.onMessage(TaskMessageListener.java:21) 	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1591) 	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1510) 	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1498) 	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1489) 	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1433) 	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:975) 	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:921) 	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:83) 	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1296) 	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1202) 	at java.lang.Thread.run(Thread.java:748) 2024-09-22 20:36:31|taskMessageListenerContainer-3|com.qsl.rabbit.listener.TaskMessageListener|INFO|20|消费者接收到消息：ab 2024-09-22 20:36:31|taskMessageListenerContainer-3|com.qsl.rabbit.listener.TaskMessageListener|INFO|37|处理任务：ab 2024-09-22 20:36:31|taskMessageListenerContainer-3|org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer|ERROR|1268|Consumer thread error, thread abort. java.lang.OutOfMemoryError: 模拟内存溢出 	at com.qsl.rabbit.listener.TaskMessageListener.handleTask(TaskMessageListener.java:40) 	at com.qsl.rabbit.listener.TaskMessageListener.onMessage(TaskMessageListener.java:21) 	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1591) 	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1510) 	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1498) 	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1489) 	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1433) 	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:975) 	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:921) 	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:83) 	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1296) 	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1202) 	at java.lang.Thread.run(Thread.java:748) 2024-09-22 20:36:31|taskMessageListenerContainer-1|org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer|ERROR|1415|Stopping container from aborted consumer 2024-09-22 20:36:31|taskMessageListenerContainer-1|org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer|INFO|646|Waiting for workers to finish. 2024-09-22 20:36:31|taskMessageListenerContainer-1|org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer|INFO|649|Successfully waited for workers to finish.  

可以看到，除了我们的业务日志，还有 spring 的日志；从日志可以看出，消息一共被消费了 3 次，但无一例外，都消费失败了，每次失败日志都包括

Consumer thread error, thread abort.

Stopping container from aborted consumer

我们再去看下队列消费者情况