Spring Boot中异步线程池@Async详解

Posted 2023-03-14

参考技术A

1、消息队列MQ

2、线程池处理。

我们来看看Spring框架中如何去使用线程池来完成异步操作，以及分析背后的原理。

在Spring4中，Spring中引入了一个新的注解@Async，这个注解让我们在使用Spring完成异步操作变得非常方便。

Spring异步线程池的接口类，其实质是java.util.concurrent.Executor

Spring 已经实现的异常线程池：

Spring中用@Async注解标记的方法，称为异步方法。在spring boot应用中使用@Async很简单：

1、调用异步方法类上或者启动类加上注解@EnableAsync

2、在需要被异步调用的方法外加上@Async

3、所使用的@Async注解方法的类对象应该是Spring容器管理的bean对象；

启动类加上注解@EnableAsync：

在需要被异步调用的方法外加上@Async，同时类AsyncService加上注解@Service或者@Component，使其对象成为Spring容器管理的bean对象；

这里需要注意的是：

1、同一个类里面调用异步方法不生效：原因默认类内的方法调用不会被aop拦截，即调用方和被调用方是在同一个类中，是无法产生切面的，该对象没有被Spring容器管理。即@Async方法不生效。

解决办法：如果要使同一个类中的方法之间调用也被拦截，需要使用spring容器中的实例对象，而不是使用默认的this，因为通过bean实例的调用才会被spring的aop拦截

本例使用方法：AsyncService asyncService = context.getBean(AsyncService.class); 然后使用这个引用调用本地的方法即可达到被拦截的目的

备注：这种方法只能拦截protected，default，public方法，private方法无法拦截。这个是spring aop的一个机制。

2、如果不自定义异步方法的线程池默认使用SimpleAsyncTaskExecutor。SimpleAsyncTaskExecutor：不是真的线程池，这个类不重用线程，每次调用都会创建一个新的线程。并发大的时候会产生严重的性能问题。

3、异步方法返回类型只能有两种：void和java.util.concurrent.Future。

1）当返回类型为void的时候，方法调用过程产生的异常不会抛到调用者层面，

可以通过注AsyncUncaughtExceptionHandler来捕获此类异常

2）当返回类型为Future的时候，方法调用过程产生的异常会抛到调用者层面

在Spring Boot主类中定义一个线程池，public Executor taskExecutor() 方法用于自定义自己的线程池，线程池前缀”taskExecutor-”。如果不定义，则使用系统默认的线程池。

上面我们通过ThreadPoolTaskExecutor创建了一个线程池，同时设置了如下参数：

核心线程数10：线程池创建时初始化的线程数

最大线程数20：线程池最大的线程数，只有在缓冲队列满了之后才会申请超过核心线程数的线程

缓冲队列200：用来缓冲执行任务的队列

允许线程的空闲时间60秒：超过了核心线程数之外的线程，在空闲时间到达之后会被销毁

线程池名的前缀：设置好了之后可以方便我们定位处理任务所在的线程池

线程池对拒绝任务的处理策略：此处采用了CallerRunsPolicy策略，当线程池没有处理能力的时候，该策略会直接在execute方法的调用线程中运行被拒绝的任务；如果执行程序已被关闭，则会丢弃该任务

设置线程池关闭的时候等待所有任务都完成再继续销毁其他的Bean

设置线程池中任务的等待时间，如果超过这个时候还没有销毁就强制销毁，以确保应用最后能够被关闭，而不是阻塞住

也可以单独类来配置线程池：

只需要在@Async注解中指定线程池名即可

Bean文件配置: spring_async.xml
1. 线程的前缀为xmlExecutor
2. 启动异步线程池配置

启动类导入xml文件：

线 程池参数说明

1. ‘id’ ： 线程名称的前缀

2. ‘pool-size’：线程池的大小。支持范围”min-max”和固定值（此时线程池core和max sizes相同）

3. ‘queue-capacity’ ：排队队列长度

4. ‘rejection-policy’: 对方拒绝的任务处理策略

5. ‘keep-alive’ ： 线程保护时间（单位秒）

上面也提到：在调用方法时，可能出现方法中抛出异常的情况。在异步中主要有有两种异常处理方法：

a) 、一种是在调用future的get时捕获异常;

b)、 在异常方法中直接捕获异常

实现AsyncConfigurer接口对异常线程池更加细粒度的控制
a) 创建线程自己的线程池
b) 对void方法抛出的异常处理的类AsyncUncaughtExceptionHandler

上面也提到：如果不自定义异步方法的线程池默认使用SimpleAsyncTaskExecutor。SimpleAsyncTaskExecutor：不是真的线程池，这个类不重用线程，每次调用都会创建一个新的线程。并发大的时候会产生严重的性能问题。

一般的错误OOM：OutOfMemoryError：unable to create new native thread，创建线程数量太多，占用内存过大.

解决办法：一般最好使用自定义线程池，做一些特殊策略, 比如自定义拒绝策略，如果队列满了，则拒绝处理该任务。

原文链接：https://blog.csdn.net/hguisu/article/details/106671893

Spring Boot系列二 Spring @Async异步线程池用法总结

1. TaskExecutor

Spring异步线程池的接口类，其实质是java.util.concurrent.Executor

Spring 已经实现的异常线程池：
1. SimpleAsyncTaskExecutor：不是真的线程池，这个类不重用线程，每次调用都会创建一个新的线程。
2. SyncTaskExecutor：这个类没有实现异步调用，只是一个同步操作。只适用于不需要多线程的地方
3. ConcurrentTaskExecutor：Executor的适配类，不推荐使用。如果ThreadPoolTaskExecutor不满足要求时，才用考虑使用这个类
4. SimpleThreadPoolTaskExecutor：是Quartz的SimpleThreadPool的类。线程池同时被quartz和非quartz使用，才需要使用此类
5. ThreadPoolTaskExecutor ：最常使用，推荐。 其实质是对java.util.concurrent.ThreadPoolExecutor的包装

2. @Async

spring对过@Async定义异步任务

异步的方法有3种
1. 最简单的异步调用，返回值为void
2. 带参数的异步调用 异步方法可以传入参数
3. 异常调用返回Future

详细见代码：

@Component
public class AsyncDemo {
    private static final Logger log = LoggerFactory.getLogger(AsyncDemo.class);

    /**
     * 最简单的异步调用，返回值为void
     */
    @Async
    public void asyncInvokeSimplest() {
        log.info("asyncSimplest");
    }

    /**
     * 带参数的异步调用 异步方法可以传入参数
     * 
     * @param s
     */
    @Async
    public void asyncInvokeWithParameter(String s) {
        log.info("asyncInvokeWithParameter, parementer={}", s);
    }

    /**
     * 异常调用返回Future
     * 
     * @param i
     * @return
     */
    @Async
    public Future<String> asyncInvokeReturnFuture(int i) {
        log.info("asyncInvokeReturnFuture, parementer={}", i);
        Future<String> future;
        try {
            Thread.sleep(1000 * 1);
            future = new AsyncResult<String>("success:" + i);
        } catch (InterruptedException e) {
            future = new AsyncResult<String>("error");
        }
        return future;
    }

}

以上的异步方法和普通的方法调用相同

asyncDemo.asyncInvokeSimplest();
asyncDemo.asyncInvokeWithException("test");
Future<String> future = asyncDemo.asyncInvokeReturnFuture(100);
System.out.println(future.get());

3. Spring 开启异步配置

Spring有两种方法启动配置
1. 注解
2. XML

3.1 通过注解实现

要启动异常方法还需要以下配置
1. @EnableAsync 此注解开户异步调用功能
2. public AsyncTaskExecutor taskExecutor() 方法自定义自己的线程池，线程池前缀”Anno-Executor”。如果不定义，则使用系统默认的线程池。

@SpringBootApplication
@EnableAsync // 启动异步调用
public class AsyncApplicationWithAnnotation {
    private static final Logger log = LoggerFactory.getLogger(AsyncApplicationWithAnnotation.class);

    /**
     * 自定义异步线程池
     * @return
     */
    @Bean
    public AsyncTaskExecutor taskExecutor() {  
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); 
        executor.setThreadNamePrefix("Anno-Executor");
        executor.setMaxPoolSize(10);  

        // 设置拒绝策略
        executor.setRejectedExecutionHandler(new RejectedExecutionHandler() {
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                // .....
            }
        });
        // 使用预定义的异常处理类
        // executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

        return executor;  
    } 

    public static void main(String[] args) {
        log.info("Start AsyncApplication.. ");
        SpringApplication.run(AsyncApplicationWithAnnotation.class, args);
    }
}

以上的异常方法和普通的方法调用相同

@RunWith(SpringRunner.class)
@SpringBootTest(classes=AsyncApplicationWithAnnotation.class)
public class AsyncApplicationWithAnnotationTests {
    @Autowired
    private AsyncDemo asyncDemo;

    @Test
    public void contextLoads() throws InterruptedException, ExecutionException {
        asyncDemo.asyncInvokeSimplest();
        asyncDemo.asyncInvokeWithParameter("test");
        Future<String> future = asyncDemo.asyncInvokeReturnFuture(100);
        System.out.println(future.get());
    }
}

执行测试用例，输出内容如下：
可以看出主线程的名称为main; 异步方法则使用 Anno-Executor1，可见异常线程池起作用了

2017-03-28 20:00:07.731  INFO 5144 --- [ Anno-Executor1] c.hry.spring.async.annotation.AsyncDemo  : asyncSimplest
2017-03-28 20:00:07.732  INFO 5144 --- [ Anno-Executor1] c.hry.spring.async.annotation.AsyncDemo  : asyncInvokeWithParameter, parementer=test
2017-03-28 20:00:07.751  INFO 5144 --- [ Anno-Executor1] c.hry.spring.async.annotation.AsyncDemo  : asyncInvokeReturnFuture, parementer=100
success:100
2017-03-28 20:00:08.757  INFO 5144 --- [       Thread-2] s.c.a.AnnotationConfigApplicationContext : Closing org.springframework.context.annotation.AnnotationConfigApplicationContext@47af7f3d: startup date [Tue Mar 28 20:00:06 CST 2017]; root of context hierarchy

3.2 通过XML实现

Bean文件配置: spring_async.xml
1. 线程的前缀为xmlExecutor
2. 启动异步线程池配置

    <!-- 等价于 @EnableAsync， executor指定线程池 -->
    <task:annotation-driven executor="xmlExecutor"/>
    <!-- id指定线程池产生线程名称的前缀 -->
    <task:executor
        id="xmlExecutor"
        pool-size="5-25"
        queue-capacity="100"
        keep-alive="120"
        rejection-policy="CALLER_RUNS"/>

线程池参数说明
1. ‘id’ ： 线程的名称的前缀
2. ‘pool-size’：线程池的大小。支持范围”min-max”和固定值（此时线程池core和max sizes相同）
3. ‘queue-capacity’ ：排队队列长度
○ The main idea is that when a task is submitted, the executor will first try to use a free thread if the number of active threads is currently less than the core size.
○ If the core size has been reached, then the task will be added to the queue as long as its capacity has not yet been reached.
○ Only then, if the queue’s capacity has been reached, will the executor create a new thread beyond the core size.
○ If the max size has also been reached, then the executor will reject the task.
○ By default, the queue is unbounded, but this is rarely the desired configuration because it can lead to OutOfMemoryErrors if enough tasks are added to that queue while all pool threads are busy.
4. ‘rejection-policy’: 对拒绝的任务处理策略
○ In the default ThreadPoolExecutor.AbortPolicy, the handler throws a runtime RejectedExecutionException upon rejection.
○ In ThreadPoolExecutor.CallerRunsPolicy, the thread that invokes execute itself runs the task. This provides a simple feedback control mechanism that will slow down the rate that new tasks are submitted.
○ In ThreadPoolExecutor.DiscardPolicy, a task that cannot be executed is simply dropped.
○ In ThreadPoolExecutor.DiscardOldestPolicy, if the executor is not shut down, the task at the head of the work queue is dropped, and then execution is retried (which can fail again, causing this to be repeated.)
5. ‘keep-alive’ ： 线程保活时间（单位秒）
setting determines the time limit (in seconds) for which threads may remain idle before being terminated. If there are more than the core number of threads currently in the pool, after waiting this amount of time without processing a task, excess threads will get terminated. A time value of zero will cause excess threads to terminate immediately after executing a task without remaining follow-up work in the task queue()

异步线程池

@SpringBootApplication
@ImportResource("classpath:/async/spring_async.xml")
public class AsyncApplicationWithXML {
    private static final Logger log = LoggerFactory.getLogger(AsyncApplicationWithXML.class);

    public static void main(String[] args) {
        log.info("Start AsyncApplication.. ");
        SpringApplication.run(AsyncApplicationWithXML.class, args);
    }
}

测试用例

@RunWith(SpringRunner.class)
@SpringBootTest(classes=AsyncApplicationWithXML.class)
public class AsyncApplicationWithXMLTest {
    @Autowired
    private AsyncDemo asyncDemo;

    @Test
    public void contextLoads() throws InterruptedException, ExecutionException {
        asyncDemo.asyncInvokeSimplest();
        asyncDemo.asyncInvokeWithParameter("test");
        Future<String> future = asyncDemo.asyncInvokeReturnFuture(100);
        System.out.println(future.get());
    }
}

运行测试用例，输出内容如下：
可以看出主线程的名称为main; 异步方法则使用 xmlExecutor-x，可见异常线程池起作用了

2017-03-28 20:12:10.540  INFO 12948 --- [           main] c.h.s.a.xml.AsyncApplicationWithXMLTest  : Started AsyncApplicationWithXMLTest in 1.441 seconds (JVM running for 2.201)
2017-03-28 20:12:10.718  INFO 12948 --- [  xmlExecutor-2] com.hry.spring.async.xml.AsyncDemo       : asyncInvokeWithParameter, parementer=test
2017-03-28 20:12:10.721  INFO 12948 --- [  xmlExecutor-1] com.hry.spring.async.xml.AsyncDemo       : asyncSimplest
2017-03-28 20:12:10.722  INFO 12948 --- [  xmlExecutor-3] com.hry.spring.async.xml.AsyncDemo       : asyncInvokeReturnFuture, parementer=100
success:100
2017-03-28 20:12:11.729  INFO 12948 --- [       Thread-2] s.c.a.AnnotationConfigApplicationContext : Closing org.springframework.context.annotation.AnnotationConfigApplicationContext@71809907: startup date [Tue Mar 28 20:12:09 CST 2017]; root of context hierarchy

4. 对异步方法的异常处理

在调用方法时，可能出现方法中抛出异常的情况。在异步中主要有有两种异常处理方法：
1. 对于方法返回值是Futrue的异步方法: a) 一种是在调用future的get时捕获异常; b) 在异常方法中直接捕获异常
2. 对于返回值是void的异步方法：通过AsyncUncaughtExceptionHandler处理异常

AsyncExceptionDemo：

@Component
public class AsyncExceptionDemo {
    private static final Logger log = LoggerFactory.getLogger(AsyncExceptionDemo.class);

    /**
     * 最简单的异步调用，返回值为void
     */
    @Async
    public void asyncInvokeSimplest() {
        log.info("asyncSimplest");
    }

    /**
     * 带参数的异步调用 异步方法可以传入参数
     *  对于返回值是void，异常会被AsyncUncaughtExceptionHandler处理掉
     * @param s
     */
    @Async
    public void asyncInvokeWithException(String s) {
        log.info("asyncInvokeWithParameter, parementer={}", s);
        throw new IllegalArgumentException(s);
    }

    /**
     * 异常调用返回Future
     *  对于返回值是Future，不会被AsyncUncaughtExceptionHandler处理，需要我们在方法中捕获异常并处理
     *  或者在调用方在调用Futrue.get时捕获异常进行处理
     * 
     * @param i
     * @return
     */
    @Async
    public Future<String> asyncInvokeReturnFuture(int i) {
        log.info("asyncInvokeReturnFuture, parementer={}", i);
        Future<String> future;
        try {
            Thread.sleep(1000 * 1);
            future = new AsyncResult<String>("success:" + i);
            throw new IllegalArgumentException("a");
        } catch (InterruptedException e) {
            future = new AsyncResult<String>("error");
        } catch(IllegalArgumentException e){
            future = new AsyncResult<String>("error-IllegalArgumentException");
        }
        return future;
    }

}

实现AsyncConfigurer接口对异常线程池更加细粒度的控制
a) 创建线程自己的线程池
b) 对void方法抛出的异常处理的类AsyncUncaughtExceptionHandler

/**
 * 通过实现AsyncConfigurer自定义异常线程池，包含异常处理
 * 
 * @author hry
 *
 */
@Service
public class MyAsyncConfigurer implements AsyncConfigurer{
    private static final Logger log = LoggerFactory.getLogger(MyAsyncConfigurer.class);

    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor threadPool = new ThreadPoolTaskExecutor();  
        threadPool.setCorePoolSize(1);  
        threadPool.setMaxPoolSize(1);  
        threadPool.setWaitForTasksToCompleteOnShutdown(true);  
        threadPool.setAwaitTerminationSeconds(60 * 15);  
        threadPool.setThreadNamePrefix("MyAsync-");
        threadPool.initialize();
        return threadPool;  
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
         return new MyAsyncExceptionHandler();  
    }

    /**
     * 自定义异常处理类
     * @author hry
     *
     */
    class MyAsyncExceptionHandler implements AsyncUncaughtExceptionHandler {  

        @Override  
        public void handleUncaughtException(Throwable throwable, Method method, Object... obj) {  
            log.info("Exception message - " + throwable.getMessage());  
            log.info("Method name - " + method.getName());  
            for (Object param : obj) {  
                log.info("Parameter value - " + param);  
            }  
        }  

    } 

}

@SpringBootApplication
@EnableAsync // 启动异步调用
public class AsyncApplicationWithAsyncConfigurer {
    private static final Logger log = LoggerFactory.getLogger(AsyncApplicationWithAsyncConfigurer.class);

    public static void main(String[] args) {
        log.info("Start AsyncApplication.. ");
        SpringApplication.run(AsyncApplicationWithAsyncConfigurer.class, args);
    }


}

测试代码

@RunWith(SpringRunner.class)
@SpringBootTest(classes=AsyncApplicationWithAsyncConfigurer.class)
public class AsyncApplicationWithAsyncConfigurerTests {
    @Autowired
    private AsyncExceptionDemo asyncDemo;

    @Test
    public void contextLoads() throws InterruptedException, ExecutionException {
        asyncDemo.asyncInvokeSimplest();
        asyncDemo.asyncInvokeWithException("test");
        Future<String> future = asyncDemo.asyncInvokeReturnFuture(100);
        System.out.println(future.get());
    }
}

运行测试用例
MyAsyncConfigurer 捕获AsyncExceptionDemo 对象在调用asyncInvokeWithException的异常

2017-04-02 16:01:45.591  INFO 11152 --- [      MyAsync-1] c.h.s.a.exception.AsyncExceptionDemo     : asyncSimplest
2017-04-02 16:01:45.605  INFO 11152 --- [      MyAsync-1] c.h.s.a.exception.AsyncExceptionDemo     : asyncInvokeWithParameter, parementer=test
2017-04-02 16:01:45.608  INFO 11152 --- [      MyAsync-1] c.h.s.async.exception.MyAsyncConfigurer  : Exception message - test
2017-04-02 16:01:45.608  INFO 11152 --- [      MyAsync-1] c.h.s.async.exception.MyAsyncConfigurer  : Method name - asyncInvokeWithException
2017-04-02 16:01:45.608  INFO 11152 --- [      MyAsync-1] c.h.s.async.exception.MyAsyncConfigurer  : Parameter value - test
2017-04-02 16:01:45.608  INFO 11152 --- [      MyAsync-1] c.h.s.a.exception.AsyncExceptionDemo     : asyncInvokeReturnFuture, parementer=100
error-IllegalArgumentException
2017-04-02 16:01:46.656  INFO 11152 --- [       Thread-2] s.c.a.AnnotationConfigApplicationContext : Closing org.springframework.context.annotation.AnnotationConfigApplicationContext@47af7f3d: startup date [Sun Apr 02 16:01:44 CST 2017]; root of context hierarchy

5. 源码地址

代码的GITHUB地址