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

本文轉載自查看原文 2017-12-16 23:13 10131 spring-boot @Async 的使用、自定義Executor的配置方法/ Spring Boot

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注解：

[java] view plain copy

@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：

[java] view plain copy

/**
* 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、萬事俱備，開始測試：

[java] view plain copy

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.");
}
}

測試結果：

[plain] view plain copy

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注解搬到這里來：

[java] view plain copy

@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的使用：

[java] view plain copy

/**
* 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、測試（測試用例不變）結果：

[plain] view plain copy

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，再寫異步方法，最后調用就可以了。

關於多線程中還有許多問題，如線程同步等就需要在寫代碼時多注意多思考了。另外，合理配置線程池參數也很重要

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 spring boot使用自定義配置的線程池執行Async異步任務 Spring Boot 2 使用自定義配置 spring-boot自定義異常返回更加靈活的參數校驗，Spring-boot自定義參數校驗注解 Spring Boot 屬性配置&自定義屬性配置 Spring Boot2.0自定義配置文件使用 Spring Boot使用@ConfigurationProperties 讀取自定義的properties的方法 Spring Boot 自定義 Shiro 過濾器，無法使用 @Autowired 解決方法 spring boot + thymeleaf +security自定義規則的簡單使用 spring boot使用自定義參數解析器