spring-boot @Async 的使用、自定义Executor的配置方法

2022-12-22 22:12:05

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地址

简单几步，实现异步新线程调用。

1、在主类中添加@EnableAsync注解：

@SpringBootApplication
@EnableScheduling
@EnableAsync
public class MySpringBootApplication {
private static Logger logger = LoggerFactory.getLogger(MySpringBootApplication.class);
public static void main(String[] args) {
SpringApplication.run(MySpringBootApplication.class, args);
logger.info("My Spring Boot Application Started");
}
}

2、创建一个AsyncTask类，在里面添加两个用@Async注解的task：

/**
* Asynchronous Tasks
* @author Xu
*
*/
@Component
public class AsyncTask {
protected final Logger logger = LoggerFactory.getLogger(this.getClass());
@Async
public Future<String> doTask1() throws InterruptedException{
logger.info("Task1 started.");
long start = System.currentTimeMillis();
Thread.sleep(5000);
long end = System.currentTimeMillis();
logger.info("Task1 finished, time elapsed: {} ms.", end-start);
return new AsyncResult<>("Task1 accomplished!");
}
@Async
public Future<String> doTask2() throws InterruptedException{
logger.info("Task2 started.");
long start = System.currentTimeMillis();
Thread.sleep(3000);
long end = System.currentTimeMillis();
logger.info("Task2 finished, time elapsed: {} ms.", end-start);
return new AsyncResult<>("Task2 accomplished!");
}
}

3、万事俱备，开始测试：

public class TaskTests extends BasicUtClass{
@Autowired
private AsyncTask asyncTask;
@Test
public void AsyncTaskTest() throws InterruptedException, ExecutionException {
Future<String> task1 = asyncTask.doTask1();
Future<String> task2 = asyncTask.doTask2();
while(true) {
if(task1.isDone() && task2.isDone()) {
logger.info("Task1 result: {}", task1.get());
logger.info("Task2 result: {}", task2.get());
break;
}
Thread.sleep(1000);
}
logger.info("All tasks finished.");
}
}

测试结果：

2016-12-13 11:12:24,850:INFO main (AsyncExecutionAspectSupport.java:245) - No TaskExecutor bean found for async processing
2016-12-13 11:12:24,864:INFO SimpleAsyncTaskExecutor-1 (AsyncTask.java:22) - Task1 started.
2016-12-13 11:12:24,865:INFO SimpleAsyncTaskExecutor-2 (AsyncTask.java:34) - Task2 started.
2016-12-13 11:12:27,869:INFO SimpleAsyncTaskExecutor-2 (AsyncTask.java:39) - Task2 finished, time elapsed: 3001 ms.
2016-12-13 11:12:29,866:INFO SimpleAsyncTaskExecutor-1 (AsyncTask.java:27) - Task1 finished, time elapsed: 5001 ms.
2016-12-13 11:12:30,853:INFO main (TaskTests.java:23) - Task1 result: Task1 accomplished!
2016-12-13 11:12:30,853:INFO main (TaskTests.java:24) - Task2 result: Task2 accomplished!
2016-12-13 11:12:30,854:INFO main (TaskTests.java:30) - All tasks finished.

可以看到，没有自定义的Executor，所以使用缺省的TaskExecutor 。

前面是最简单的使用方法。如果想使用自定义的Executor，可以按照如下几步来：

1、新建一个Executor配置类，顺便把@EnableAsync注解搬到这里来：

@Configuration
@EnableAsync
public class ExecutorConfig {
/** Set the ThreadPoolExecutor's core pool size. */
private int corePoolSize = 10;
/** Set the ThreadPoolExecutor's maximum pool size. */
private int maxPoolSize = 200;
/** Set the capacity for the ThreadPoolExecutor's BlockingQueue. */
private int queueCapacity = 10;
@Bean
public Executor mySimpleAsync() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(corePoolSize);
executor.setMaxPoolSize(maxPoolSize);
executor.setQueueCapacity(queueCapacity);
executor.setThreadNamePrefix("MySimpleExecutor-");
executor.initialize();
return executor;
}
@Bean
public Executor myAsync() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(corePoolSize);
executor.setMaxPoolSize(maxPoolSize);
executor.setQueueCapacity(queueCapacity);
executor.setThreadNamePrefix("MyExecutor-");
// rejection-policy：当pool已经达到max size的时候，如何处理新任务
// CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
executor.initialize();
return executor;
}
}

这里定义了两个不同的Executor，第二个重新设置了pool已经达到max size时候的处理方法；同时指定了线程名字的前缀。

2、自定义Executor的使用：

/**
* Asynchronous Tasks
* @author Xu
*
*/
@Component
public class AsyncTask {
protected final Logger logger = LoggerFactory.getLogger(this.getClass());
@Async("mySimpleAsync")
public Future<String> doTask1() throws InterruptedException{
logger.info("Task1 started.");
long start = System.currentTimeMillis();
Thread.sleep(5000);
long end = System.currentTimeMillis();
logger.info("Task1 finished, time elapsed: {} ms.", end-start);
return new AsyncResult<>("Task1 accomplished!");
}
@Async("myAsync")
public Future<String> doTask2() throws InterruptedException{
logger.info("Task2 started.");
long start = System.currentTimeMillis();
Thread.sleep(3000);
long end = System.currentTimeMillis();
logger.info("Task2 finished, time elapsed: {} ms.", end-start);
return new AsyncResult<>("Task2 accomplished!");
}
}

就是把上面自定义Executor的类名，放进@Async注解中。

3、测试（测试用例不变）结果：

2016-12-13 10:57:11,998:INFO MySimpleExecutor-1 (AsyncTask.java:22) - Task1 started.
2016-12-13 10:57:12,001:INFO MyExecutor-1 (AsyncTask.java:34) - Task2 started.
2016-12-13 10:57:15,007:INFO MyExecutor-1 (AsyncTask.java:39) - Task2 finished, time elapsed: 3000 ms.
2016-12-13 10:57:16,999:INFO MySimpleExecutor-1 (AsyncTask.java:27) - Task1 finished, time elapsed: 5001 ms.
2016-12-13 10:57:17,994:INFO main (TaskTests.java:23) - Task1 result: Task1 accomplished!
2016-12-13 10:57:17,994:INFO main (TaskTests.java:24) - Task2 result: Task2 accomplished!
2016-12-13 10:57:17,994:INFO main (TaskTests.java:30) - All tasks finished.
2016-12-13 10:57:18,064 Thread-3 WARN Unable to register Log4j shutdown hook because JVM is shutting down. Using SimpleLogger

可见，线程名字的前缀变了，两个task使用了不同的线程池了。

源代码：https://github.com/xujijun/my-spring-boot

引言：Spring作为容器为我们托管对象，但是有时我们需要多线程执行任务，那么我们该如何配置呢？

解决：利用java的线程池Executor执行任务

步骤

1.配置TaskExecutor

这里直接将线程池注入

CorePoolSize代表执行任务的线程数量

public class TaskExecutorConfig implements AsyncConfigurer{//实现AsyncConfigurer接口

    @Bean

    public Executor getAsyncExecutor() {//实现AsyncConfigurer接口并重写getAsyncExecutor方法，并返回一个ThreadPoolTaskExecutor，这样我们就获得了一个基于线程池TaskExecutor

         ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();

            taskExecutor.setCorePoolSize(5);

            taskExecutor.setMaxPoolSize(10);

            taskExecutor.setQueueCapacity(25);

            taskExecutor.initialize();

            return taskExecutor;

    }

    @Override

    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {

        return null;

    }

}

这里我们得到了一个基于java的线程池Executer的线程池，然后设置了部分参数，返回了一个实例

2.编写我们需要执行的任务，并注明service

@Service

@Slf4j

public class AsyncTaskService {

  @Async

  public void dataTranslate(int i)

  {
log.info("启动了线程"+i);

  }

}

这里用日志打印

3.将线程池对象注入，并调用任务service。

最后在application中开始异步支持@EnableAsync

调用结果：可以看到，是不同的线程执行了打印任务，而且根据cpu时间片，抢占，可以看到线程执行顺序也发生了变化，说明是异步执行

总结：Spring Boot对多线程的支持和Spring没什么两样，就是需要先配置线程池，然后注入bean，再写异步方法，最后调用就可以了。

关于多线程中还有许多问题，如线程同步等就需要在写代码时多注意多思考了。另外，合理配置线程池参数也很重要

码农公寓