跳到主要內容

CompletableFuture - Avoid blocking inside CompletableFuture chains. It costs.

Problem

如下圖所示,使用ExecutorService最害怕的就是遇到Blocking的操作,這將會降低整個ExecutorService的生產力。

(本篇圖片都來自於link)

How to resolve?

面對這問題,作者建議針對Blocking操作可以在發送後,透過Aux Thread執行;中間透過completableFuture的compose串接;Aux Thread執行完畢後,會將結果丟給串接的completableFuture,並讓工作繼續回到原本的ExecutorService執行。如下圖所示:

我嘗試使用Java8實作,範例程式碼如下:

private static int cpu_num = Runtime.getRuntime().availableProcessors();
private static ExecutorService es = Executors.newFixedThreadPool(cpu_num, 
			CustomThreadFactory.create("cpu_bound_thread_"));
 
public static class BlockingJob {
	public Response invoke() {
		try {
			dumpCurrentThreadName("before blocking job");
			Thread.sleep(2*1000);
		} catch( Exception e ) {
			e.printStackTrace();
		} finally {
			dumpCurrentThreadName("after blocking job");
		}
		return new Response();
	}
}
 
private static <T> CompletableFuture<T> launchTaskWithAuxThread(Supplier<T> supplier){
	ExecutorService auxEs = 
			Executors.newSingleThreadExecutor(CustomThreadFactory.create("aux_blocking_thread_"));
	CompletableFuture<T> task = CompletableFuture
			.supplyAsync(supplier, auxEs);
	auxEs.shutdown();
	return task;
}
 
@Test
public void testSendAsync(){
	CompletableFuture<Response> sendAsync = CompletableFuture.supplyAsync(()->{
		dumpCurrentThreadName("supplyAsync");
		return launchTaskWithAuxThread(()->new BlockingJob().invoke());
	}, es)
	.thenCompose((CompletableFuture<Response> responseReceived)->{
		dumpCurrentThreadName("thenCompose");
		return responseReceived;
	})
	.thenApplyAsync((Response x)->{
		dumpCurrentThreadName("thenApply");
		return x;
	}, es);
 
	sendAsync.join();
}

實作特別需要注意的兩個部分是,

  • Aux Thread儘量別使用與原本相同的ExecutorService。假如你使用的是固定大小的ExecutorService,可能會造成Deadlock;假如你是使用CachedExecutorService,就必須要注意Thread數量是否會造成OOM。
  • thenCompose之後的動作必須使用Async,否則會是用Aux Thread去執行後續操作,無法將CPU bound工作集中。(這部分可能會根據實作會有差異)

輸出如下:

cpu_bound_thread_1: supplyAsync
cpu_bound_thread_1: thenCompose
aux_blocking_thread_1: before blocking job
aux_blocking_thread_1: after blocking job
cpu_bound_thread_2: thenApply

從結果可以發現,除了blocking的操作外,其餘的操作都是跑在我特定的ExecutorService中。我的想法是,

  • 如果此ExecutorService專門用於CPU Bound工作,這做法可分離出Blocking操作避免影響到ThreadPool的生產力。
  • 範例中針對Blocking操作是建立單一Thread的ExecutorService,但如果會有IO最大限制的情況下,這部分需要控管數量。
  • 這方法雖然有額外的thread context switch,但處於Blocking狀態的Thread並不會與CPU搶資源,因此工作忙碌時所帶來的效能優勢應可掩蓋此缺點。

Note. 我的範例程式只是實現的方法之一,應要根據需求做調整。 例如原始投影片中,是透過responseReceived.completeAsync讓工作回到compose之後繼續往下處理。

Reference

留言

這個網誌中的熱門文章

PostgreSQL - Unattended installation on windows

Introduction 要將別人軟體包裝到自己軟體中,不可或缺的東西就是Unattended installation。以Unattended installation來說,我們可以選擇透過Installer的silent mode安裝,也可以透過把目標軟體做成portable的版本。本篇文章分享這兩種方法,教導大家如何將PostgreSQL透過Unattended installation方式安裝到目標系統成為service。 Note. 本篇以PostgreSQL 10.7為例。 Install with installer Tips 安裝程式或反安裝程式的參數,除了可以直接上官網搜尋Installation User Guide以外,也可以直接使用help參數查詢: postgresql- 10.7 - 2 -windows-x64.exe --help Windows安裝程式主要有EnterpriseDB與BigSQL兩種。BigSQL版本安裝元件是透過網路下載且支援參數不如EnterpriseDB版本多,以我們需求來說,我們傾向於使用EnterpriseDB版本。接下來分享給大家安裝與反安裝方法。 Installation @ echo off set INSTALL_DIR =C:\postgres10 set INSTALLER =postgresql- 10.7 - 2 -windows-x64.exe   rem options for installation set SSMDB_SERVICE =postgresql- 10 set MODE =--unattendedmodeui none --mode unattended   set DB_PASSWD =--superpassword postgres set DB_PORT =--serverport 5432   set SERVICE_NAME =--servicename % SSMDB_SERVICE %   set PREFIX =--prefix "%INSTALL_DIR%" set DATA_DIR =--datadir "%INSTALL_DIR%\data"   set OPTIONS =

How to install RIDE on Windows?

Introduction 多年沒在Windows上開發RobotFramework,趁著這次整理一下RIDE安裝方法。 目前RIDE最新版本與Python對應版本如下: (3.6 < python <= 3.11) Install current released version (2.0.8.1) with: pip install -U robotframework-ride 安裝Python 直接到Python官網找尋最新的3.11版本,我使用3.11.9: link 。安裝就是一直下一步而已。 安裝wxPython 每次安裝RIDE最困難的都是wxPython。看了一下 官網 描述,我就姑且相信一下: 接著進入下 載頁面 就有安裝教學。基本上就是到Python目錄下的Scripts直接執行以下command: pip install -U wxPython 安裝RIDE 接著就如RIDE官網所說,執行以下command: pip install -U robotframework-ride 啟動RIDE 直接在相同目錄下執行ride就可以啟動了,你也可以直接在桌面建ride連結,加快下次啟動時間。 沒想到這次這麼順利就安裝完成了。因為我是使用java去啟動robot framework,就不特別講要怎麼使用pip安裝robot framework了。

Hello World!

即將要搬家,因此舊網頁內容將慢慢轉移至Blogger。 如果要存取舊網頁,可以使用以下連結: https://wiki.tonylin.idv.tw/dokuwiki/doku.php