线上异常处理(一): 线程堆积导致OOM, RabbitMQ的Connection中自动创建线程池

谁践踏了优雅 2021-12-02 01:04 514阅读 0赞

# 说明 #

本篇博文主要记录之前线上项目由于线程数过多导致内存溢出后，事故原因的分析排查过程。项目背景是其中使用了公司封装的管理类来操作RabbitMQ。

# 正文 #

## 初步猜想 ##

项目出现oom后发现是线程数过多导致内存泄露，快速进行了重启。重启后，仍**发现线程数在不断地增长**。 查看代码发现，在从mq获取到消息后会进行http请求调用其他服务。查看工具类，发现每次请求都会新建CloseableHttpClient对象，请求完成后调用close()方法关闭连接。所以**怀疑是连接没有关闭导致线程泄露**。

对此，我们修改了工具类，创建CloseableHttpClient单例对象，并且使用连接池管理，自定义线程保活策略。设置最大连接数为50，默认最大单路径连接数为5，连接最长存活10s。

private static final int MAX_TOTAL = 50;
    
    private static final int DEFAULT_MAX_PERROUTE = 5;
    
    private static final int KEEP_ALIVE_TIMEOUT = 10 * 1000;
    
    private static CloseableHttpClient HTTP_CLIENT;
    
    static {
        PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager();
        connectionManager.setMaxTotal(MAX_TOTAL);
        connectionManager.setDefaultMaxPerRoute(DEFAULT_MAX_PERROUTE);
        ConnectionKeepAliveStrategy strategy = new ConnectionKeepAliveStrategy() {
            @Override
            public long getKeepAliveDuration(HttpResponse response, HttpContext context) {
                HeaderElementIterator it = new BasicHeaderElementIterator
                        (response.headerIterator(HTTP.CONN_KEEP_ALIVE));
                while (it.hasNext()) {
                    HeaderElement he = it.nextElement();
                    String param = he.getName();
                    String value = he.getValue();
                    if (value != null && param.equalsIgnoreCase
                            ("timeout")) {
                        return Long.parseLong(value) * 1000;
                    }
                }
                return KEEP_ALIVE_TIMEOUT;
            }
        };
        client = HttpClients.custom().setKeepAliveStrategy(strategy).setConnectionManager(connectionManager).build();
    }

在测试中发现，**线程数仍在不断增长，之前的问题并没有得到解决。所以我们使用jstack命令观察线程堆栈信息。**

--------------------

## jstack 分析线程堆栈信息 ##

使用jstack得到线程堆栈信息后发现，**有大量的线程处于TIMED\_WAITING状态，并且线程名称都是pool-x-thread-x的形式，并且线程编号在不断增长**。从名称上看，这部分线程是由线程池创建，并且由于pool-x的不同，断定是**由多个不同的线程池创建**。

但是，查看项目代码后发现，项目中并没有使用多个线程池，并且在使用线程池时指定了创建线程名称的格式。

通过jstack无法得到线程池对象，所以我们使用jmap命令分析堆内存信息。

--------------------

## jmap 生成dump文件 ##

使用命令 **jmap -dump:format=b,file=heapdump.phrof pid 生成堆转储快照dump文件。** 生成快照文件后，使用MemoryAnalyze工具查看堆中对象。

通过dump文件，我们现在堆中有多个AMQConnection对象，在该对象下发现了创建名为pool-x线程的线程池对象。

在项目中我们使用了RabbitMQ，所以通过阅读源码，找见创建线程池的代码。

--------------------

## 阅读源码 ##

在AMQConnection类源码中，有三个ExecutorService实例变量。分别是：

private final ExecutorService consumerWorkServiceExecutor;
    private final ScheduledExecutorService heartbeatExecutor;
    private final ExecutorService shutdownExecutor;

在start方法中，使用initializeConsumerWorkService()方法，创建consumerWorkServiceExecutor对象。创建ConsumerWorkService对象。

private void initializeConsumerWorkService() {
        this._workService  = new ConsumerWorkService(consumerWorkServiceExecutor, threadFactory, shutdownTimeout);
    }

**在ConsumerWorkService的构造函数中，创建了newFixedThreadPool线程池，线程数量为处理器核心数的2倍。**

private static final int DEFAULT_NUM_THREADS = Runtime.getRuntime().availableProcessors() * 2;
    
    public ConsumerWorkService(ExecutorService executor, ThreadFactory threadFactory, int shutdownTimeout) {
        this.privateExecutor = (executor == null);
        this.executor = (executor == null) ? Executors.newFixedThreadPool(DEFAULT_NUM_THREADS, threadFactory)
                                           : executor;
        this.workPool = new WorkPool<Channel, Runnable>();
        this.shutdownTimeout = shutdownTimeout;
    }

而创建线程时，使用的是AMQConnection类中创建的默认的threadFactory:

private ThreadFactory threadFactory = Executors.defaultThreadFactory();

默认线程工厂创建的线程名称格式为pool-x-thread-y:

DefaultThreadFactory() {
        SecurityManager s = System.getSecurityManager();
        group = (s != null) ? s.getThreadGroup() :
                              Thread.currentThread().getThreadGroup();
        namePrefix = "pool-" +
                      poolNumber.getAndIncrement() +
                     "-thread-";
    }

至此，我们已经清楚内存中大量的线程是由AMQConnection中的CosumerService对象中的线程池创建的。

**为什么会出现多个AMQConnection对象？**  
上面提到在该项目中，使用了公司封装的管理类，通过阅读管理类的源码，发现为了实现队列的并发消费，**封装类中使用了线程封闭的方式创建Consumer对象，每个Consumer对象都是由单独的Connection创建，一个Consumer对应Connection**。当创建多个Consumer时会创建多个线程池，内存中线程数量飙升导致OOM。

最后，**通过降低消费者数量减少线程的创建，避免由于线程数量过多导致OOM。**

--------------------

## 总结 ##

本篇博文主要是梳理了在排查线上项目内存溢出时的处理过程，由初步猜想到使用工具分析，再到源码阅读，一步步发现问题的原因。从中学习到要熟悉jvm调优命令的使用，学会阅读源码，分析源码。

**思考：为什么要使用线程封闭Connection创建Consumer来达到并发消费？**  
通过之前对RabbitMQ的学习，了解到每个Connection都是一个TCP连接，为了避免过多的TCP连接，AMQP协议提供了Channel，Channel共享Connection。线程封闭Connection，失去了Channel的意义，可以在一个Connection上创建多个Channel，进而创建多个Consumer，也可以达到并发消费，提升吞吐量。

通过之前的学习，我们还了解到为提升消息的吞吐量，我们可以设置prefetchCount来指定每个消费者可以携带最大未确认消息的数量，还可以修改消息的确认方式来提升吞吐量。接下来我将继续翻译学RabbitMQ文档中有关消息消费的相关内容。