A服务作为生产者,每天发送上千万的mq消息,每一个消息包含500个用户ids数据。B服务作为消费者,接受MQ消息并通过http调用第三方请求进行业务处理,消费组启用了rabbitmq的多线程消费组,一个实例并发40个mq消费者线程,每个线程一次获取10个消息进行消费。
Mq消费者配置如下:
- # mq配置
- rabbitmq:
- connection-timeout: 15000
- cache:
- channel:
- size: 200
- # 消息发送到rabbitmq broker cluster需要回调
- publisher-confirms: true
- # 交换机将消息投递至队列失败时需要回调
- publisher-returns: true
- listener:
- # 手动确认消息已被消费
- simple:
- acknowledge-mode: manual
- # consumer的并发数
- concurrency: 40
- max-concurrency: 50
- # 每个消息者每次取10条
- prefetch: 10
Mq挤压消息如下
找出rabbitmq挤压的消息,在本地模拟消费,找出没有进行消息确认的原因,通过rabbitmq控制台的Get messages功能
复制payload的消息进行base64转码,转出来的消息是乱码不完整的,怀疑
是rabbitmq还结合了其他加密处理,放弃这种排查思路
rabbitmq的unack消息挤压,那就是消费者没有进行ack确认,怀疑消费者代码有异常导致没能执行到ack的代码。
查询服务器日志,没发现有报错的日志,梳理业务代码,消费者使用了spring aop around机制进行消息确认,所以不管代码有没有报错,按理说都会手动进行mq消息ack确认
消费组实例数量符合服务器大小配置,因此服务器应用没有宕机
使用IBM的TMDA工具进行分析线程堆栈,工具下载地址
TMDA工具下载地址
TMDA工具简介
TMDA分析线程堆栈结果如下
通过分析图,看到大量park线程,确实是符合现状,应用的线程挂起了
通过stack深度高到底排序,业务代码存在线程等待情况,具体代码CountDownLatch.await
通过上图stack提示,找到关联的业务代码
伪代码如下:
- // new一个CompletableFuture
- public CompletableFuture
httpCall(String tokenData){ - CompletableFuture
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;
- });
熟练CompletableFuture的使用,要看源码的实现(实现原理cas + 多个future采用入stack,每次把前一个future的结果作为参数传入下一个future去执行)
使用多线程需要考虑异常、超时等情况
熟练使用jvm stack分析工具
CompletableFuture流程图
CompletableFuture参考文章如下