线程堵塞(挂起)导致消息队列消息挤压

news2024/10/6 14:35:52

一、背景

A服务作为生产者，每天发送上百万的mq消息，每一个消息包含500个用户id数据。

B服务作为消费者，接受MQ消息并通过http调用第三方请求进行业务处理，消费组使用了rabbitmq

的多线程消费组，一个实例并发40个mq消费者线程，每个线程一次获取10个消息进行消费

Mq消费者配置如下
  rabbitmq:
    connection-timeout: 2000
    cache:
      channel:
        size: 100
    # 消息发到rabbitmq-broker-cluster回调
    publisher-confirms: true
    # 交换机将消息投递至队列失败时回调
    publisher-returns: true
    listener:
      # 手动确认消息已被消费
      simple:
        acknowledge-mode: manual
        # consumer的并发数
        concurrency: 40
        max-concurrency: 50
        # 每个消息者每次取多少数据
        prefetch: 10

Mq挤压消息如下

二、排查

排查思路1

找出rabbitmq挤压的消息，在本地模拟消费，找出没有进行消息确认的原因，通过rabbitmq控制台的Get messages功能

复制payload的消息进行base64转码，转出来的消息是乱码不完整的，怀疑

是rabbitmq还结合了其他加密处理，放弃这种排查思路

排查思路2

rabbitmq的unack消息挤压，那就是消费者没有进行ack确认，怀疑消费者代码有异常导致没能执行到ack的代码。

查询服务器日志，没发现有报错的日志，

梳理业务代码，消费者使用了spring aop around机制进行消息确认，所以不管代码有没有报错，按理说都会手动进行mq消息ack确认

排查思路3

服务器应用没有挂掉

消费组数量是符合配置大小

怀疑是线程挂起，不执行消息确认代码导致消息的挤压

进行排查应用线程，通过jstack命令导出jvm线程

使用IBM的TMDA工具进行分析线程堆栈

工具下载地址：https://publib.boulder.ibm.com/httpserv/cookbook/Major_Tools-IBM_Thread_and_Monitor_Dump_Analyzer_TMDA.html

三、线程堆栈分析

TMDA分析如下

通过分析图，看到大量park线程，确实是符合现状，应用的线程挂起了

通过stack深度高到底排序，业务代码存在线程等待情况，具体代码CountDownLatch.await

四、业务代码分析

通过上图stack提示，找到关联的业务代码

伪代码如下：

// new一个CompletableFuture
public CompletableFuture<Integer> httpCall(String tokenData){
        CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
            long time = 3000L;
            try {
                Thread.sleep(time);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return Integer.parseInt(tokenData);
        });
        return completableFuture;
    }

httpCall(tokenData).whenCompleteAsync((returnValue, ex)->{
		// do business
		// ex.getMessage()
		// 其中ex对象为空，使用ex.getMessage()报了空指针，导致没能执行如下的countDown
		countDownLatch.countDown();
	})

消费者通过http调用第三方服务，为了提高并发，使用了多线程，每一组(数十个为一组)http请求批次调用完成后再把请求响应结果存入数据库。

主线程使用了countDownLatch.await进行等待。

其中whenCompleteAsync方法存在空指针问题，导致没能执行如下的countDown。

这里有人会问不是说日志没有空指针异常吗

对，子线程报了空指针，因为CompletableFuture执行每次都是new 一个新的CompletableFuture对象，并把结果作为下一个CompletableFuture执行的入参，通过伪代码可以发现，执行whenCompleteAsync后，没有新的CompletableFuture方法执行，所以异常没有抛出来，使得排查变得困难

五、解决

因为存在whenCompleteAsync报错的情况，添加多一个新的处理方法，捕获异常也进行

countDown的操作

代码如下：

httpCall(tokenData).whenCompleteAsync((returnValue, ex)->{
		// do business
		// ex.getMessage()
		// 其中ex对象为空，使用ex.getMessage()报了空指针，导致没能执行如下的countDown
		countDownLatch.countDown();
	}).exceptionally(e ->{
                log.info("exceptionally捕获到异常，tokenData={}, e={}", tokenData, e.getMessage());
                countDownLatch.countDown();
                return null;
            })