转

springboot-@Async默认线程池导致OOM问题

本文链接：https://blog.csdn.net/ignorewho/article/details/85603920

内存溢出的三种类型：
1.第一种OutOfMemoryError： PermGen space，发生这种问题的原意是程序中使用了大量的jar或class
2.第二种OutOfMemoryError： Java heap space，发生这种问题的原因是java虚拟机创建的对象太多
3.第三种OutOfMemoryError：unable to create new native thread，创建线程数量太多，占用内存过大

初步分析：

代码分析：

借此机会也学习了下SimpleAsyncTaskExecutor源码，总结如下：
1.SimpleAsyncTaskExecutor提供了限流机制，通过concurrencyLimit属性来控制开关，当concurrencyLimit>=0时开启限流机制，默认关闭限流机制即concurrencyLimit=-1，当关闭情况下，会不断创建新的线程来处理任务，核心代码如下：

public void execute(Runnable task, long startTimeout) {
   Assert.notNull(task, "Runnable must not be null");
   Runnable taskToUse = (this.taskDecorator != null ? this.taskDecorator.decorate(task) : task);
   //判断是否开启限流机制
   if (isThrottleActive() && startTimeout > TIMEOUT_IMMEDIATE) {
      //执行前置操作，进行限流
      this.concurrencyThrottle.beforeAccess();
      //执行完线程任务，会执行后置操作concurrencyThrottle.afterAccess()，配合进行限流
      doExecute(new ConcurrencyThrottlingRunnable(taskToUse));
   }
   else {
      doExecute(taskToUse);
   }
}

2.SimpleAsyncTaskExecutor限流实现
首先任务进来，会循环判断当前执行线程数是否超过concurrencyLimit，如果超了，则当前线程调用wait方法，释放monitor对象锁，进入等待

protected void beforeAccess() {
    if (this.concurrencyLimit == NO_CONCURRENCY) {
        throw new IllegalStateException(
                "Currently no invocations allowed - concurrency limit set to NO_CONCURRENCY");
    }
    if (this.concurrencyLimit > 0) {
        boolean debug = logger.isDebugEnabled();
        synchronized (this.monitor) {
            boolean interrupted = false;
            while (this.concurrencyCount >= this.concurrencyLimit) {
                if (interrupted) {
                    throw new IllegalStateException("Thread was interrupted while waiting for invocation access, " +
                            "but concurrency limit still does not allow for entering");
                }
                if (debug) {
                    logger.debug("Concurrency count " + this.concurrencyCount +
                            " has reached limit " + this.concurrencyLimit + " - blocking");
                }
                try {
                    this.monitor.wait();
                }
                catch (InterruptedException ex) {
                    // Re-interrupt current thread, to allow other threads to react.
                    Thread.currentThread().interrupt();
                    interrupted = true;
                }
            }
            if (debug) {
                logger.debug("Entering throttle at concurrency count " + this.concurrencyCount);
            }
            this.concurrencyCount++;
        }
    }
}

2.SimpleAsyncTaskExecutor限流实现：首先任务进来，会循环判断当前执行线程数是否超过concurrencyLimit，如果超了，则当前线程调用wait方法，释放monitor对象锁，进入等待状态。

protected void beforeAccess() {
    if (this.concurrencyLimit == NO_CONCURRENCY) {
        throw new IllegalStateException(
                "Currently no invocations allowed - concurrency limit set to NO_CONCURRENCY");
    }
    if (this.concurrencyLimit > 0) {
        boolean debug = logger.isDebugEnabled();
        synchronized (this.monitor) {
            boolean interrupted = false;
            while (this.concurrencyCount >= this.concurrencyLimit) {
                if (interrupted) {
                    throw new IllegalStateException("Thread was interrupted while waiting for invocation access, " +
                            "but concurrency limit still does not allow for entering");
                }
                if (debug) {
                    logger.debug("Concurrency count " + this.concurrencyCount +
                            " has reached limit " + this.concurrencyLimit + " - blocking");
                }
                try {
                    this.monitor.wait();
                }
                catch (InterruptedException ex) {
                    // Re-interrupt current thread, to allow other threads to react.
                    Thread.currentThread().interrupt();
                    interrupted = true;
                }
            }
            if (debug) {
                logger.debug("Entering throttle at concurrency count " + this.concurrencyCount);
            }
            this.concurrencyCount++;
        }
    }
}

线程任务执行完毕后，当前执行线程数会减一，会调用monitor对象的notify方法，唤醒等待状态下的线程，等待状态下的线程会竞争monitor锁，竞争到，会继续执行线程任务。

protected void afterAccess() {
    if (this.concurrencyLimit >= 0) {
        synchronized (this.monitor) {
            this.concurrencyCount--;
            if (logger.isDebugEnabled()) {
                logger.debug("Returning from throttle at concurrency count " + this.concurrencyCount);
            }
            this.monitor.notify();
        }
    }
}

虽然看了源码了解了SimpleAsyncTaskExecutor有限流机制，实践出真知，我们还是测试下：
一、测试未开启限流机制下，我们启动20个线程去调用异步方法，查看Java VisualVM工具如下：

二、测试开启限流机制，开启限流机制的代码如下:

@Configuration
@EnableAsync
public class AsyncCommonConfig extends AsyncConfigurerSupport {
    @Override
    public Executor getAsyncExecutor() {
        SimpleAsyncTaskExecutor executor = new SimpleAsyncTaskExecutor();
        //设置允许同时执行的线程数为10
 executor.setConcurrencyLimit(10);
        return executor;
    }
}

同样，我们启动20个线程去调用异步方法，查看Java VisualVM工具如下：

最终解决办法：
1.自定义线程池，使用LinkedBlockingQueue阻塞队列来限定线程池的上限
2.定义拒绝策略，如果队列满了，则拒绝处理该任务，打印日志，代码如下：

public class AsyncConfig implements AsyncConfigurer{
    private Logger logger = LogManager.getLogger();

    @Value("${thread.pool.corePoolSize:10}")
    private int corePoolSize;

    @Value("${thread.pool.maxPoolSize:20}")
    private int maxPoolSize;

    @Value("${thread.pool.keepAliveSeconds:4}")
    private int keepAliveSeconds;

    @Value("${thread.pool.queueCapacity:512}")
    private int queueCapacity;

    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(corePoolSize);
        executor.setMaxPoolSize(maxPoolSize);
        executor.setKeepAliveSeconds(keepAliveSeconds);
        executor.setQueueCapacity(queueCapacity);
        executor.setRejectedExecutionHandler((Runnable r, ThreadPoolExecutor exe) -> {
                logger.warn("当前任务线程池队列已满.");
        });
        executor.initialize();
        return executor;
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return new AsyncUncaughtExceptionHandler() {
            @Override
            public void handleUncaughtException(Throwable ex , Method method , Object... params) {
                logger.error("线程池执行任务发生未知异常.", ex);
            }
        };
    }
}

</article>