不恰当使用线程池处理 MQ 消息引起的故障

标签: 杂记 MQ 线程池 | 发表时间:2014-08-30 16:56 | 作者:coderbee
出处:http://coderbee.net

现状

业务部门反应网站访问特别慢,负责运维监控的同事说MQ消息队列积压了,中间件的说应用服务器内存占用很高,GC 一直回收不了内存,GC 线程占了近 100% 的 CPU,其他的基本上都在等待,数据库很正常,完全没压力。没啥办法,线程、堆 dump 出来后,重启吧,然后应用又正常了。

分析

这种故障之前其实也碰到过了,分析了当时 dump 出来的堆后发现,处理 MQ 消息的线程池的队列长度达百万级别,占用了超过 1.3G 内存,这些内存都是没法回收的。

程序的实现目前是这样的:关联系统把消息推送到 MQ 上,我们再从 MQ 上拉消息下来处理;每种类型的消息都有一个线程负责从 MQ 上拉消息,拉下来后封装成线程池的任务提交给相应的线程池去执行。代码可以简化为:

  package net.coderbee.mq.demo;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class MQListener {
     public ExecutorService executor = Executors.newFixedThreadPool(8);

     public void onMessage(final Object message) {
          executor.execute(new Runnable() {
               @Override
               public void run() {
                    // 耗时且复杂的消息处理逻辑
                    complicateHanlde(message);
               }
          });
     }

     private void complicateHanlde(Object message) {
     }
}

这个实现就是导致故障的根源, Executors.newFixedThreadPool(8) 创建的线程池的任务队列是无边界的:

  public static ExecutorService newFixedThreadPool(int nThreads) {
     return new ThreadPoolExecutor(nThreads, nThreads,
                                          0L, TimeUnit.MILLISECONDS,
                                          new LinkedBlockingQueue<Runnable>());
}

当时是关联系统出故障了,他们恢复后,往 MQ 里狂推消息,我们系统里面的 MQListener 不断地从 MQ 拉消息下来,直接塞进线程池里,由于线程池处理消息的速度远远慢于消息进入的速度,所以线程池的队列不断增长,直到把所有的堆内存都占用了,这时不断引发 FullGC,但每次 FullGC 都没法回收到内存,应用也就挂死在那了。

之前那次故障也是线程池队列积压导致的,引起的原因是消息处理逻辑调用了外部接口,由于外部接口的响应非常慢,严重拖慢了消息的处理进度,改成异步调用之后好了些。但问题的根源并没有解决,就像昨天关联系统狂推消息后,我们的系统还是挂了。

解决方法

我的思路其实很简单,MQ 是用来系统间解耦的,也是一个缓冲,目前的实现是把处理消息的线程池又用作一个 MQ 了,消息不能不受控地进入线程池的任务队列,所以,要换成使用定长的阻塞队列,队列满了就暂停拉取消息。把线程池替换成:

  private int nThreads = 8;
private int MAX_QUEUQ_SIZE = 2000;
private ExecutorService executor = new ThreadPoolExecutor(nThreads,
          nThreads, 0L, TimeUnit.MILLISECONDS,
          new ArrayBlockingQueue<Runnable>(MAX_QUEUQ_SIZE),
          new ThreadPoolExecutor.CallerRunsPolicy());

把线程池队列满的时候直接让调用者(也就是 MQListener)执行任务,这样即延缓了消息拉取的速度,当 MQListener 再去拉取消息时,发现线程池有空间时可以提交到线程池,让线程池的工作线程去处理,它继续保持拉取速度。

这样既控制了线程池占用的内存,又可以让消息处理线程池处理不过来时多一个线程处理消息。

由于上面的代码采用调用者执行的方式,那么要考虑消息处理的顺序问题,比如一个订单的处理可能有多个步骤,对应多条 MQ 消息,那么要考虑这些步骤如果乱序了是否可以接受,因为第3步骤的处理消息可能被 MQListener 处理了,而第2步的处理消息还积压在线程池里。

相关 [线程池 mq 消息] 推荐:

不恰当使用线程池处理 MQ 消息引起的故障

- - 码蜂笔记
业务部门反应网站访问特别慢,负责运维监控的同事说MQ消息队列积压了,中间件的说应用服务器内存占用很高,GC 一直回收不了内存,GC 线程占了近 100% 的 CPU,其他的基本上都在等待,数据库很正常,完全没压力. 没啥办法,线程、堆 dump 出来后,重启吧,然后应用又正常了. 这种故障之前其实也碰到过了,分析了当时 dump 出来的堆后发现,处理 MQ 消息的线程池的队列长度达百万级别,占用了超过 1.3G 内存,这些内存都是没法回收的.

消息队列mq的3个使用场景

- - 互联网 - ITeye博客
2、tomcat controller取到请求后向rocketmq 发送一个msg,将msg id返回给app,同时在redis里缓存msg状态为init(设置定时时间,时间到后清除). 3、client(app/h5/小程序) 通过msg id,定时向server获取msg处理状态. (init时 画圈,没有时返回繁忙,fail时返回处理失败).

GitHub - 事务最终一致性 - MQ保证消息一致性

- -
使用rabbitMq来简单实现分布式事务的最终一致性 版本如下:. 目录 名称 访问地址. gateway 网关+路由. rabbitMq -> 注册中心 -> 网关 -> 订单 -> 支付. 3、发送消息(采用确认模式). 4、mq收到消息之后给生产端一个确认消息. 5、生产端监听这个确认消息.

使用MQ的时候,怎么确保消息100%不丢失?

- - 掘金后端本月最热
面试官在面试候选人时,如果发现候选人的简历中写了在项目中使用了 MQ 技术(如 Kafka、RabbitMQ、RocketMQ),基本都会抛出一个问题: 在使用 MQ 的时候,怎么确保消息 100% 不丢失. 这个问题在实际工作中很常见,既能考察候选者对于 MQ 中间件技术的掌握程度,又能很好地区分候选人的能力水平.

闲扯kafka mq

- - 开源软件 - ITeye博客
本文主要讲解关于kafka mq的设计思想及个人理解. 关于kafka的详细信息,大家可以参考官网的文献 http://kafka.apache.org/documentation.html这是一篇相当不错的文章,值得仔细研读. 第一个问题:消息队列(Message Queue)是干嘛用的. 首先,要对消息队列有一个基本的理解.

[MQ]关于ActiveMQ的配置

- - 企业架构 - ITeye博客
  目前常用的消息队列组建无非就是MSMQ和ActiveMQ,至于他们的异同,这里不想做过多的比较. 简单来说,MSMQ内置于微软操作系统之中,在部署上包含一个隐性条件:Server需要是微软操作系统. (对于这点我并去调研过MSMQ是否可以部署在非微软系统,比如:Linux,只是拍脑袋想了想,感觉上是不可以).

与MQ通讯的完整JAVA程序

- - 编程语言 - ITeye博客
        本文实例是基于 WebSphere MQ中将消息发送至远程队列的配置的基础上的,且如果要能正常运行并发送、接收消息,还需要在两个队列管理器(QM_ORANGE和QM_APPLE)上做如下配置或修改.         1.创建名称为DC.SVRCONN的服务器连接通道.         2.将队列管理器的通道认证记录设置为“已禁用”.

Java线程池

- - 企业架构 - ITeye博客
线程的使用在java中占有极其重要的地位,在jdk1.4极其之前的jdk版本中,关于线程池的使用是极其简陋的. 在jdk1.5之后这一情况有了很大的改观. Jdk1.5之后加入了java.util.concurrent包,这个包中主要介绍java中线程以及线程池的使用. 为我们在开发中处理线程的问题提供了非常大的帮助.

Java 线程池

- - 编程语言 - ITeye博客
在项目中,系统启动一个新线程的成本是比较高的,因为它涉及与操作系统交互. 在这种情形下,使用线程池可以很好地提高性能,尤其是当程序中需要创建大量生存周期很短的线程时,更应该考虑使用线程池. 使用线程池可以有效地控制系统中并发线程的数量,当系统中包含大量并发线程时,会导致系统性能剧烈下降,甚至导致JVM崩溃,而线程池的最大线程数参数可以控制系统中并发线程数不超过此数.

在Tomcat 6.0下用JNDI连接IBM MQ 6.0的配置方法

- - 行业应用 - ITeye博客
假设在IBM MQ中定义的队列管理器的名为QueueManager, 端口1414,CCSID 437 ,创建名为LQ1,LQ2的队列分别用于发送和接收消息, 服务器连接通道名为SVRCONN. 确保在项目的Classpath中导入了以下的jar包:. 如果需使用spring的JmsTemplate方式来读写MQ,还需要导入.