Java服務端開發非同步化實踐的一點小結
一、背景
隨著Java 9、Spring 5以及Spring Boot 2對於響應式程式設計（Reactive Stack）的支援，Reactor模式也日漸趨於流行，雖然當前並未在業界全面落地應用，但隨著Java 8函數語言程式設計、Lambda表示式、Stream資料流式處理的深入人心，加之Spring框架對於事件驅動模型的有力支援，日後響應式程式設計的全面落地當只是時間問題。而響應式程式設計的一個很重要的特性就是它對非同步化的完美封裝使得開發人員只需要專注於資料流的處理和業務邏輯的實現而不用過於關心非同步化背後複雜的執行緒管理，但這既是響應式程式設計模式的優點也是它的缺點，導致它的學習曲線較高，程式碼也並不是那麼好理解，它是一種可以說是完全不同於我們已經習慣了的同步程式設計思維的新的程式設計模式，如果不能完全深刻地理解便不敢輕易應用於成熟的業務中，這也是導致它雖然已經出現了有一些年頭了但至今尚未能夠大規模落地實踐的一個很重要的原因。
而對於Java語言，響應式框架中最為流行的當是來源於微軟的LINQ擴充套件函式庫的RxJava了。它在微服務的服務治理元件來源於Netflix公司的服務容錯框架Hystrix的實現中得到了全面的使用，從而使其在微服務的全面落地過程中聲名鵲起，如果想要研究Hystrix的具體實現便不得不首先學習RxJava的一些基本概念、操作符和API使用。由於RxJava和響應式程式設計尚未在公司的業務中全面應用，本篇並不對它們做過多深究，這篇文章主要是從實踐的角度談談在當前的同步程式設計模式下如何結合現有在使用的技術應用非同步業務處理。
二、什麼是非同步，為何要使用非同步
同步：執行緒阻塞等待結果的響應再進行下一步的處理。這個我們已經非常熟悉了，它在我們的業務處理中佔據了主流，也符合我們的思維流程。
非同步：主執行緒不需要阻塞等待響應而能夠繼續進行後續的處理，而把這一部分需要一定耗時的業務邏輯交予子執行緒處理，在子執行緒處理完畢後再通知主執行緒或另外的執行緒結果進行後續的處理或無需任何後續處理。
從上面的描述中可以看出，同步與非同步最大的區別便在於會不會阻塞住當前執行緒，使其處於無謂的等待中。
那麼，我們為什麼要使用非同步，非同步相比於同步有什麼好處？這個問題可以使用我們在中學時代學過的著名數學家華羅庚關於燒水泡茶的兩個演算法來說明：
演算法一：
第一步、燒水；第二步、水燒開後，洗刷茶具；第三步、沏茶
演算法二：
第一步、燒水；第二步、燒水過程中，洗刷茶具；第三步、水燒開後沏茶
這兩個演算法我們一眼便能看出哪個更節省時間。其實我們的日常編碼基本也都是這樣的類似工序的安排，如何更好地安排好業務處理步驟在更快的時間內做出響應便是我們需要做的優化。上面的演算法一便是完全同步的演算法，上一步完成了才能進行下一步，這樣雖然非常有利於我們的理解和對於工序的控制，但卻不能充分利用資源浪費了很多無謂的時間在等待中什麼也沒做。而演算法二則是結合了同步和非同步，充分利用了燒水過程中的等待時間來洗刷茶具，有效地利用了時間資源，從而比演算法一要節省不少時間，能夠更快地做出響應（沏好茶）。對於有依賴關係的步驟同步不可避免，但是對於沒有任何依賴關係的步驟完全可以使用非同步進行並行處理以提高處理效率在更快的時間內做出響應。因此非同步的好處便是可以充分利用資源並行或併發處理提高處理效率節省響應時間，對應於計算機來說，便是充分利用多核CPU的處理能力在同一時間內做更多的計算，提高CPU的資源利用率，提高系統的吞吐量和降低系統的響應時間。因為我們的業務處理中，大部分的邏輯都是IO密集型的，比如存取資料庫、遠端服務呼叫（RPC或HTTP等）、讀寫檔案等，而非CPU密集型的（對於大量的資料需要CPU計算的可使用ForkJoinPool/ForkJoinTask進行任務分解來平行計算），IO密集型的任務正如上面的燒水過程一樣，需要一定的耗時，會使執行緒阻塞等待，此時它並不佔用CPU資源，因此在這個等待的過程中是浪費了CPU寶貴的資源的。如果對這樣的IO呼叫進行非同步化並行或併發處理，此時主執行緒便不會阻塞而會繼續處理後面的邏輯，從而很好地利用了CPU。而這個在網際網路環境下應對高併發請求的場景中尤為重要，是我們一切優化所努力的方向。快取、訊息佇列（MQ）等中介軟體的作用也在於此。
三、Java服務端開發中非同步化的手段
從第二部分的描述中也可以看出，非同步化的前提是要正確梳理出業務處理步驟中的依賴關係，哪些步驟是有依賴關係的，哪些步驟是沒有依賴關係的，對於有依賴關係的步驟使用同步的方式處理，對於沒有依賴關係的步驟則可以使用非同步進行並行或併發處理。當然，有依賴關係的幾個步驟組合起來在大的處理時序中可能與其他的步驟之間又是沒有依賴關係的，此時在更高一層級中大的步驟之間也可以使用非同步的方式，需要我們根據具體的業務邏輯具體問題具體分析靈活處理。
那麼，在Java服務端的開發中，都有哪些非同步化的手段呢？
1、多執行緒執行緒池（ThreadPoolExecutor）的方式
程式碼如下：
ExecutorService executor = Executors.newFixedThreadPool(5);
// 不需要關心處理結果
executor.execute(() -> System.out.println("業務處理步驟..."));
// 需要處理結果進行後續的邏輯處理
Future<String> future = executor.submit(() -> {
System.out.println("業務處理步驟...");
return "業務處理結果";
});
// 中間的其他處理步驟
// 對非同步任務的結果進行處理
try {
System.out.println("非同步任務的結果：" + future.get());
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}複製程式碼
這種方式雖然簡單明瞭也很常用，但是它的主要問題在於把業務邏輯處理和執行緒處理程式碼耦合在了一起，使得執行緒的程式碼侵入到了我們的業務中，不利於執行緒的統一管理和維護，沒有一個統一的執行緒集中管理收口很容易造成執行緒的無限制濫用，從而引起系統問題。而且，Future的使用非常不靈活，它的獲取結果的get()方法還是阻塞式的，有點雞肋，因此並不建議大量使用。當然，也可以結合Google的Guava庫提供的SettableFuture和ListenableFuture進行改造，但這畢竟需要另外的工作量，還有很多問題需要處理。
下面重點來說說在Java服務端開發非同步化實踐中個人覺得比較好的兩種方式：
2、Spring的事件機制和@Async註解非同步任務支援
結合Spring的事件機制和@Async註解非同步方法呼叫可以實現非同步化的處理，主要程式碼如下：
首先宣告開啟非同步功能支援和非同步任務執行執行緒池配置，還可以配置非同步任務異常處理器
@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {
@Bean
@Override
public Executor getAsyncExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(10);
executor.setMaxPoolSize(100);
executor.setQueueCapacity(100);
executor.setThreadNamePrefix("EventExecutor-");
return executor;
}
@Override
public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
//return new EventAsyncUncaughtExceptionHandler();
return null;
}
static class EventAsyncUncaughtExceptionHandler implements AsyncUncaughtExceptionHandler {
@Override
public void handleUncaughtException(Throwable ex, Method method, Object... params) {
}
}
}複製程式碼
基礎事件定義
public class BaseEvent extends ApplicationEvent {
private static final long serialVersionUID = 5570783330749390897L;
public BaseEvent(Object source) {
super(source);
}
}複製程式碼
事件釋出器
@Component
public class EventPublisher {
@Autowired
private ApplicationContext context;
public void publishEvent(BaseEvent event){
context.publishEvent(event);
}
}複製程式碼
具體的業務事件
@Getter
@Setter
@ToString
public class MisRecognitionEvent extends BaseEvent {
private static final long serialVersionUID = 8645189838824356680L;
private MisRecognitionInfo record;
public MisRecognitionEvent(Object source, MisRecognitionInfo record) {
super(source);
this.record = record;
}
}
複製程式碼
具體的業務程式碼中釋出可非同步處理的事件訊息
MisRecognitionInfo info = new MisRecognitionInfo(); MisRecognitionEvent misRecognitionEvent = new MisRecognitionEvent(this, info); eventPublisher.publishEvent(misRecognitionEvent);複製程式碼
事件處理器
@Component
@Slf4j
public class FaceMonitorEventListener {
@EventListener
@Async
public void handleMisRecognitionEvent(MisRecognitionEvent event) {
MisRecognitionInfo info = event.getRecord();
// 事件處理邏輯
}
}複製程式碼
上面的模式適合主流程不關心非同步流程的結果（比如記錄日誌，報警通知等），如果需要在主流程中對非同步流程的結果進行處理，可以直接使用@Async非同步方法，而不需要使用事件的機制
@Component
@Slf4j
public class AsyncRpcService {
@Async
AsyncResult<String> getData(String dataId) {
return new AsyncResult<String>("data");
}
}複製程式碼
具體的業務程式碼中獲取非同步任務的結果並做處理
AsyncResult<String> asyncRpcResult = asyncRpcService.getData("dataId");
asyncRpcResult.addCallback((result) -> System.out.println("非同步任務結果：" + result),
(ex) -> System.out.println("非同步任務異常：" + ex.getMessage()));複製程式碼
3、Java 8的CompletableFuture的非同步模式
CompletableFuture的API介紹請參考這篇文章（ Java 8 CompletableFuture 教程 ）。
Java 8的CompletableFuture擴充套件了Future的功能，提供了對非同步任務執行完成後的回撥和對多個Future的鏈式編排，從而可以實現非常豐富的非同步任務處理場景。
場景一：
單個非同步任務的處理
（1）呼叫執行緒不需要關心非同步任務的處理結果
// 使用預設執行緒池
CompletableFuture.runAsync(() -> System.out.println("沒有返回結果的非同步任務..."));
// 使用指定執行緒池
CompletableFuture.runAsync(() -> System.out.println("沒有返回結果的非同步任務..."), executor);複製程式碼
（2）對非同步任務的結果進行處理
- 處理返回結果和異常
CompletableFuture.supplyAsync(() -> {
System.out.println("非同步任務處理...");
return "非同步任務的返回結果";
}, executor).whenComplete((result, exception) -> {
if (result != null) {
System.out.println("處理非同步任務結果");
}
if (exception != null) {
System.out.println("處理非同步任務異常");
}
});複製程式碼
場景二：
- 多個沒有依賴關係的非同步任務的處理
（1）多個非同步任務都完成後的處理
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
System.out.println("非同步任務一...");
return "非同步任務一的結果";
}, executor);
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
System.out.println("非同步任務二...");
return "非同步任務二的結果";
}, executor);
CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
System.out.println("非同步任務三...");
return "非同步任務三的結果";
}, executor);
List<CompletableFuture<String>> futureList = new ArrayList<>();
futureList.add(future1);
futureList.add(future2);
futureList.add(future3);
CompletableFuture<Void> allFutures = CompletableFuture.allOf(futureList.toArray(new CompletableFuture[futureList.size()]));
// 所有非同步任務都完成後的處理
allFutures.whenComplete((result, exception) -> {
if (exception != null) {
System.out.println("其中的一個非同步任務發生了異常：" + exception.getMessage());
}
// 每個非同步任務的結果需要單獨獲取
List<String> futureResults = futureList.stream().map(future -> future.join()).collect(Collectors.toList());
System.out.println("所有非同步任務的結果集合：" + futureResults);
});複製程式碼
需要注意的是，如果其中任何一個非同步任務發生了異常，則allFutures也會以該異常作為返回的結果，後面也就獲取不到每個非同步任務的結果了，因此這種情況下最好在每個非同步任務裡處理好異常。
（2）多個非同步任務中的任何一個完成後的處理
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
System.out.println("非同步任務一...");
try {
TimeUnit.SECONDS.sleep(2);
System.out.println("非同步任務一耗時2秒鐘...");
} catch (InterruptedException e) {
e.printStackTrace();
}
return "非同步任務一的結果";
}, executor);
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
System.out.println("非同步任務二...");
try {
TimeUnit.SECONDS.sleep(1);
System.out.println("非同步任務二耗時1秒鐘...");
} catch (InterruptedException e) {
e.printStackTrace();
}
return "非同步任務二的結果";
}, executor);
CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
System.out.println("非同步任務三...");
try {
TimeUnit.SECONDS.sleep(3);
System.out.println("非同步任務三耗時3秒鐘...");
} catch (InterruptedException e) {
e.printStackTrace();
}
return "非同步任務三的結果";
}, executor);
List<CompletableFuture<String>> futureList = new ArrayList<>();
futureList.add(future1);
futureList.add(future2);
futureList.add(future3);
CompletableFuture<Object> allFutures = CompletableFuture.anyOf(futureList.toArray(new CompletableFuture[futureList.size()]));
// 所有非同步任務都完成後的處理
allFutures.whenComplete((result, exception) -> {
if (exception != null) {
System.out.println("其中的一個非同步任務發生了異常：" + exception.getMessage());
}
if (result != null) {
System.out.println("多個非同步任務最快完成的那個任務返回的結果：" + result);
}
});複製程式碼
- 多個有依賴關係的非同步任務的鏈式編排處理
CompletableFuture.supplyAsync(() -> {
System.out.println("非同步任務一...");
return "非同步任務一的返回結果";
}, executor).thenApplyAsync(result -> {
System.out.println("非同步任務二依賴於非同步任務一的結果...");
return result + ":" + "非同步任務二的返回結果";
}, executor).thenCombineAsync(CompletableFuture.supplyAsync(() -> {
System.out.println("非同步任務三...");
return "非同步任務三的返回結果";
}, executor), (result1, result2) -> {
System.out.println("非同步任務四依賴於非同步任務二和非同步任務三的結果...");
return result1 + ":" + result2 + ":" + "非同步任務四的返回結果";
}, executor).whenComplete((result, exception) -> {
if (result != null) {
System.out.println("最終結果：" + result);
}
if (exception != null) {
System.out.println("處理非同步任務異常");
}
});複製程式碼
- 對前一個任務處理結果的三種處理方式
thenApplyAsync：接收前一個任務的返回結果並重新返回一個結果
thenAcceptAsync：只接收前一個任務的返回結果
thenRunAsync：不接收前一個任務的返回結果，只在前一個任務完成後做一些處理
需要注意的是，鏈式呼叫後面一個任務執行的前提是前一個任務沒有發生異常，否則便終止了。
- 兩個非同步任務都完成後的處理
thenCombineAsync
thenAcceptBothAsync
runAfterBothAsync
- 兩個非同步任務中的任何一個完成後的處理
applyToEitherAsync
acceptEitherAsync
runAfterEitherAsync
CompletableFuture的非同步模式（***Async方法）分別提供了使用預設執行緒池和指定執行緒池兩種方式。預設執行緒池使用的是ForkJoinPool的commonPool：
private static final boolean useCommonPool =
(ForkJoinPool.getCommonPoolParallelism() > 1);
/**
* Default executor -- ForkJoinPool.commonPool() unless it cannot
* support parallelism.
*/
private static final Executor asyncPool = useCommonPool ?
ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();
/** Fallback if ForkJoinPool.commonPool() cannot support parallelism */
static final class ThreadPerTaskExecutor implements Executor {
public void execute(Runnable r) { new Thread(r).start(); }
}複製程式碼
如果當前系統不支援平行計算（多核處理器），則會為每個任務創造一個執行緒。commonPool的執行緒數則是CPU核數-1：
/**
* Creates and returns the common pool, respecting user settings
* specified via system properties.
*/
private static ForkJoinPool makeCommonPool() {
int parallelism = -1;
ForkJoinWorkerThreadFactory factory = null;
UncaughtExceptionHandler handler = null;
try {// ignore exceptions in accessing/parsing properties
String pp = System.getProperty
("java.util.concurrent.ForkJoinPool.common.parallelism");
String fp = System.getProperty
("java.util.concurrent.ForkJoinPool.common.threadFactory");
String hp = System.getProperty
("java.util.concurrent.ForkJoinPool.common.exceptionHandler");
if (pp != null)
parallelism = Integer.parseInt(pp);
if (fp != null)
factory = ((ForkJoinWorkerThreadFactory)ClassLoader.
getSystemClassLoader().loadClass(fp).newInstance());
if (hp != null)
handler = ((UncaughtExceptionHandler)ClassLoader.
getSystemClassLoader().loadClass(hp).newInstance());
} catch (Exception ignore) {
}
if (factory == null) {
if (System.getSecurityManager() == null)
factory = defaultForkJoinWorkerThreadFactory;
else // use security-managed default
factory = new InnocuousForkJoinWorkerThreadFactory();
}
if (parallelism < 0 && // default 1 less than #cores
(parallelism = Runtime.getRuntime().availableProcessors() - 1) <= 0)
parallelism = 1;
if (parallelism > MAX_CAP)
parallelism = MAX_CAP;
return new ForkJoinPool(parallelism, factory, handler, LIFO_QUEUE,
"ForkJoinPool.commonPool-worker-");
}複製程式碼
由此可見，預設執行緒池的方式適合CPU密集型的任務，而我們大多數的業務處理都是IO密集型的，因此最好使用顯式指定執行緒池的方式。
總結一下，事件和@Async註解的方式更適合非同步處理邏輯比較複雜需要與主邏輯解耦的場景，而CompletableFuture的方式則更適合簡單輕量的非同步處理邏輯或與主邏輯有較強的關聯關係的場景。
當然，在實踐中還發現有依賴於訊息中介軟體的基於訊息的釋出訂閱模式的非同步化處理方式，但由於這種方式比較重，通常用在一些特殊的跨系統的非同步解耦的場景中，並不適合在單系統應用中大量使用，在此不再詳述。
