深入淺出Java/Spring/Spring Boot異步多線程
1、Java的多線程
1.1 線程池模式
一個線程池可以維護多個線程，這些線程等待任務來進行并發處理。線程池模式避免了頻繁創建和銷毀短期任務線程，復用池中的線程從而提高了性能。線程池中的線程在處理任務時是并發進行的。
線程池（綠色方塊）/等待處理任務隊列（藍色）/處理完成任務（黃色）
該模式允許創建的線程數量及其生命周期。 我們還能夠安排任務的執行并將傳入的任務保持在隊列（Task Queue）中。
線程池數量的大小可根據程序可用的計算資源進行調整，它通常是程序的可調參數，經過調整以優化程序的性能。 確定最佳線程池大小對于優化性能至關重要。
1.2 Java的線程池ThreadPoolExecutor
Java中的ThreadPoolExecutor是一個可擴展的線程池的實現，它提供了對很多參數的微調設置。這些參數包括：
ThreadPoolExecutor的構造器：
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
RejectedExecutionHandler handler) {
//...
}
示例：
ThreadPoolExecutor executorPool = new ThreadPoolExecutor(5, 10, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(50));
corePoolSize：第1個參數“5”。JVM會為前5個任務創建線程，后續的任務會放進隊列里，直到隊列滿為止（第5個參數workQueue，50個任務）。
maximumPoolSize：第2個參數“10”。JVM最多創建10個線程，這意味著當當前正有5個線程運行著5個任務，而這時任務隊列的50個任務已滿，此時如果有一個新的任務到達隊列，JVM將創建一個新的線程，最多創建5個，直至10個。
keepAliveTime：第3個參數“3秒”，超出核心線程而小于最大線程的這些線程，在一定的空閑時間之后將被清除掉。
unit：第4個參數“秒”，keepAliveTime的時間單位。
workQueue：第5個參數“new ArrayBlockingQueue<Runnable>(50)”，任務隊列的大小。
JVM創建線程的規則如下：
- 如果線程數少于corePoolSize，創建新的線程來跑任務。
- 如何線程數等于或大于corePoolSize，將任務放進隊列。
- 如果隊列滿了，且線程數小于maximumPoolSize，創建新的線程跑任務。
- 如果隊列滿了，且線程數量大于或等于maximumPoolSize，拒絕任務。
2、Spring的多線程
Spring/Spring Boot只需要在配置類上注解“@EnableAsync”，在需要使用單獨線程的方法上使用“@Async”注解即可。Spring會自動檢索線程池的定義，可以是“org.springframework.core.task.TaskExecutor”或者是“java.util.concurrent.Executor”的名為“taskExecutor”的bean。若都未找到，則使用“org.springframework.core.task.SimpleAsyncTaskExecutor”來處理異步方法的調用。
我們最簡單可以通過自定一個名為“taskExecutor”的Bean即可。
@Configuration
@EnableAsync
class AsyncConfigurationByBean {
@Bean(name="taskExecutor")
public Executor taskExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(5);
executor.setMaxPoolSize(10);
executor.setQueueCapacity(50);
executor.setThreadNamePrefix("poolThread-");
executor.initialize();
return executor;
}
}
Spring也提供了“AsyncConfigurer”接口用來定制實現異步多線程相關的配置。
@Configuration
@EnableAsync
class AsyncConfigurationByConfigurer implements AsyncConfigurer {
@Override
public Executor getAsyncExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(5);
executor.setMaxPoolSize(10);
executor.setQueueCapacity(50);
executor.setThreadNamePrefix("poolThread-");
executor.initialize();
return executor;
}
@Override
public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
return AsyncConfigurer.super.getAsyncUncaughtExceptionHandler();
}
}
3、Spring Boot的多線程
在Spring Boot下，通過“TaskExecutionAutoConfiguration”自動配置類，Spring Boot給我們已經自動配置好了線程池，“TaskExecutionProperties”提供了相關的屬性配置。在Spring Boot下我們只需要在配置類上“@EnableAsync”，在“application.yml”上配置即可：
spring:
task:
execution:
pool:
core-size: 5
max-size: 10
queue-capacity: 50
thread-name-prefix: poolThread-
4、多線程演示
新建一個使用異步方法的類和方法：
@Service
@Slf4j
class AsyncService {
@Async
public void doSomething(Integer i){
log.info("當前是循環：" + i);
}
}
在Spring Boot入口類調用：
@SpringBootApplication
@EnableAsync
public class SpringAsyncApplication {
public static void main(String[] args) {
SpringApplication.run(SpringAsyncApplication.class, args);
}
@Bean
CommandLineRunner commandLineRunner(AsyncService asyncService){
return p -> {
for(int i = 0 ; i < 10 ; i ++){
asyncService.doSomething(i);
}
};
}
}
運行結果：
從圖中可以看出我們的核心線程數量是5，你可以按照上節的“JVM創建線程的規則”來調整核心線程、最大線程、隊列數量的來嘗試一下觀察控制臺輸出的結果。
5、多線程異步結果演示
我們有時候會在單一的請求中調用多個方法，在同步的方法里，我們都是順序執行，執行完一個再執行下一個。我們可以通過Spring的“AsyncResult”讓多個方法并發執行并聚合他們的結果，并提高性能。
我們先看一下，在同步的情況下是什么樣的：
@Service
class Service1 {
public Integer doSomething() throws InterruptedException {
Thread.sleep(2000);
log.info("在Service1中");
return 1;
}
}
@Service
class Service2 {
public Integer doSomething() throws InterruptedException {
Thread.sleep(2000);
log.info("在Service2中");
return 2 ;
}
}
@Bean
CommandLineRunner commandLineRunner(Service1 service1, Service2 service2){
return p -> {
long start = System.currentTimeMillis();
Integer first = service1.doSomething();
Integer second = service2.doSomething();
Integer sum = first + second;
long end = System.currentTimeMillis();
Long cost = end - start;
log.info("結果為:" + sum + ",耗時" + cost);
};
}
示例中，一個線程順序執行兩個方法，執行結果為：
在一個線程中
我們現在使用“@Async”使方法變成異步，且使用“AsyncResult”包裝異步結果返回。
“CompletableFuture”是Java8引入的，以提供一種編寫異步、非阻塞和多線程代碼的簡單方法。我們使用“AsyncResult”包裝返回值，并用它的“.completable()”方法獲得“CompletableFuture”對象。
@Service
@Slf4j
class Service1 {
@Async
public CompletableFuture<Integer> doSomething() throws InterruptedException {
Thread.sleep(2000);
log.info("在Service1中");
return new AsyncResult<Integer>(1).completable();
}
}
@Service
@Slf4j
class Service2 {
@Async
public CompletableFuture<Integer> doSomething() throws InterruptedException {
Thread.sleep(2000);
log.info("在Service2中");
return new AsyncResult<Integer>(2).completable();
}
}
@Bean
CommandLineRunner commandLineRunner(Service1 service1, Service2 service2){
return p -> {
long start = System.currentTimeMillis();
CompletableFuture<Integer> firstData = service1.doSomething();
CompletableFuture<Integer> secondData = service2.doSomething();
CompletableFuture<Integer> mergeResult = firstData.thenCompose(
firstValue -> secondData.thenApply(
secondValue -> firstValue + secondValue
)
);
long end = System.currentTimeMillis();
Long cost = end - start;
log.info("結果為:" + mergeResult.get() + ",耗時" + cost);
};
}
本文轉載自今日「愛科學的衛斯理」，可以通過以下二維碼關注。轉載本文請聯系愛科學的衛斯理。
