Dubbo異步化實踐
作者：jackyjin
我們使用CompletableFuture將Dubbo接口進行了異步化，同時利用CompletableFuture的異步回調能力，減少了服務依賴之間的阻塞，增加了dubbo線程的處理請求的能力，同時利用CompletableFuture傳入的業務線程提高了服務器CPU資源的利用率，用更少的硬件資源可以處理更多的請求，為公司的降本增效貢獻了一小份力量。
1、背景
從Apach Dubbo的官網了解到從 2.7.0 版本開始，Dubbo 的所有異步編程接口開始以CompletableFuture為基礎，Dubbo接口異步化能夠極大地提高接口性能，降低接口依賴調用之間的阻塞，同時了解到我們公司大部分應用使用的是同步rpc，在公司降本增效的大背景下，我們選擇了在客服機器人組對Dubbo異步化進行落地實踐，實踐下來發現Dubbo異步化對接口性能提升了50%，涉及異步化的應用服務器縮減了1/3，接下來主要為大家分享一下實踐的經驗以及異步化提升的效果。
2、Dubbo異步化實現方式
通過CompletableFuture可以將復雜的業務邏輯從Dubbo線程池（大小默認200）切換到用戶自定義的業務線程來執行，提升Dubbo線程池請求的處理能力，同時增加自定義業務線程池，提升服務器的資源利用率。接下來我們來看下CompletableFuture怎么異步化Dubbo接口以及其原理。
2.1 接口改造方式
getRecommendContent為老的方法，asyncGetRecommendContent為新添加的異步方法；老的方法保留，兼容沒有升級的調用方；添加新的異步方法，返回值使用CompletableFuture進行包裝：
public interface RecommendAtConnectApi {
Result<RecommendAtConnectRes> getRecommendContent(RecommendAtConnectReq request);
CompletableFuture<Result<RecommendAtConnectRes>> asyncGetRecommendContent(RecommendAtConnectReq request);
}2.2 future使用方式
下面先介紹幾種常用的使用方式：
- future的結果獲取到時轉化處理（thenApply）
CompletableFuture<String> cFuture = cAsyncService.asyncSayHello(name);
CompletableFuture<DataDTO> finalFuture = cFuture.thenApply(c -> new DataDTO());
return finalFuture;- 多個future組合轉化（thenCombine），超過2個可使用allOf，后面實踐有使用到
CompletableFuture<String> cFuture = cAsyncService.asyncSayHello(name);
CompletableFuture<String> dFuture = dAsyncService.asyncSayHello(name);
CompletableFuture<DataDTO> allFuture = cFuture.thenCombine(dFuture, (c, d) -> new DataDTO());
return allFuture;- 多個future前后依賴（thenCompose）
CompletableFuture<Optional<RecommendAtConnectDto>> taskEngineFuture = pushGsTaskEngineHandler.asyncPushHandler(connectRequest);
CompletableFuture<Optional<RecommendAtConnectDto>> refundFuture = getNextFuture(taskEngineFuture, connectRequest, unused ->pushLogisticsRefundHandler.asyncPushHandler(connectRequest));
return refundFuture;
//回調工具方法
public static CompletableFuture<Optional<RecommendAtConnectDto>> getNextFuture(CompletableFuture<Optional<RecommendAtConnectDto>> beforeFuture,
RecommendAtConnectRequest request,
Function<RecommendAtConnectRequest, CompletableFuture<Optional<RecommendAtConnectDto>>> function) {
return beforeFuture.thenCompose(recommendAtConnectDto -> {
if (!recommendAtConnectDto.isPresent()) {
return function.apply(request);
}
return beforeFuture;
});
}還有很多其他的使用方式這里就不再一一介紹，大家感興趣了可以去看下官方文檔https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html
2.3 CompletableFuture原理
//CompletableFuture源碼
volatile Object result; // Either the result or boxed AltResult
volatile Completion stack; // Top of Treiber stack of dependent actionsCompletableFuture有兩個非常重要的屬性result和stack，result是future中的結果，stack是future獲取到結果時回調的函數動作存儲的棧，stack是一個Completion，Completion中有指向下一個Completion的指針。
thenApply
//thenApply源碼
private <V> CompletableFuture<V> uniApplyStage(
Executor e, Function<? super T,? extends V> f) {
if (f == null) throw new NullPointerException();
CompletableFuture<V> d = new CompletableFuture<V>();
if (e != null || !d.uniApply(this, f, null)) {
//生成當前future的依賴
UniApply<T,V> c = new UniApply<T,V>(e, d, this, f);
//CAS操作壓入棧中
push(c);
//嘗試運行一下
c.tryFire(SYNC);
}
return d;
}thenApply的原理比較簡單在調用的時候會將回調的邏輯生成UniApply壓入棧中，UniApply中包含了返回的future和當前的feture，等到當前future有結果返回時，會回調執行棧中的函數f。
thenCombine
//thenCombine源碼
private <U,V> CompletableFuture<V> biApplyStage(
Executor e, CompletionStage<U> o,
BiFunction<? super T,? super U,? extends V> f) {
CompletableFuture<U> b;
if (f == null || (b = o.toCompletableFuture()) == null)
throw new NullPointerException();
CompletableFuture<V> d = new CompletableFuture<V>();
if (e != null || !d.biApply(this, b, f, null)) {
//生成二元依賴的BiCompletion
BiApply<T,U,V> c = new BiApply<T,U,V>(e, d, this, b, f);
//將其壓入當前和組合的future棧中
bipush(b, c);
c.tryFire(SYNC);
}
return d;
}thenCombine依賴兩個future，返回一個新的future，當依賴的兩個future都有結果返回之后，回調傳入的函數動作。
thenCompose
//thenCompose源碼
private <V> CompletableFuture<V> uniComposeStage(
Executor e, Function<? super T, ? extends CompletionStage<V>> f) {
if (f == null) throw new NullPointerException();
Object r; Throwable x;
//如果線程池為空且當前future已經有結果
if (e == null && (r = result) != null) {
// try to return function result directly
if (r instanceof AltResult) {
if ((x = ((AltResult)r).ex) != null) {
return new CompletableFuture<V>(encodeThrowable(x, r));
}
r = null;
}
try {
@SuppressWarnings("unchecked") T t = (T) r;
//將當前處理結果作為f的輸入，并執行f得到新的future g
CompletableFuture<V> g = f.apply(t).toCompletableFuture();
Object s = g.result;
//如果已經有結果直接返回
if (s != null)
return new CompletableFuture<V>(encodeRelay(s));
//new一個返回的future
CompletableFuture<V> d = new CompletableFuture<V>();
//生成一個元依賴的UniCompletion
UniRelay<V> copy = new UniRelay<V>(d, g);
//將其壓入g的棧中
g.push(copy);
copy.tryFire(SYNC);
return d;
} catch (Throwable ex) {
return new CompletableFuture<V>(encodeThrowable(ex));
}
}
//如果當前結果為空，則直接生成當前feture的依賴，壓入棧中
CompletableFuture<V> d = new CompletableFuture<V>();
UniCompose<T,V> c = new UniCompose<T,V>(e, d, this, f);
push(c);
c.tryFire(SYNC);
return d;
}CompletableFuture底層借助了魔法類Unsafe的相關CAS方法，除了get或join阻塞之外，其他方法都實現了無鎖操作。
3、實踐經驗
3.1 機器人場景選擇
這次實踐主要選擇了機器人的3個場景進行改造：訂單詳情頁和聊天頁猜你想問以及輸入聯想。選擇這3個場景的原因如下：
- 接口qps高，異步化ROI高
- 大量調用外部接口，屬于IO密集型場景，異步化提升效果明顯
- 出于安全和穩定性的考慮，機器人核心的對話接口不受這3個接口異步化的影響
3.2 最佳實踐
3.2.1 梳理接口的先后依賴關系
不管是新的功能的開發還是老的代碼的改造這一步都至關重要，我們可以像梳理電路圖一樣梳理接口之間的先后依賴關系，將并行關系和串行關系梳理出來，筆者在實踐之后才明白這個道理，希望這份經驗能幫助大家少走一些彎路：
- 圖中每個CF為接口或者service返回的CompletableFuture
- CF1、CF2和CF3同一層的代表它們是并行的關系，CF2和CF4前后代表它們是依賴的關系
- 最后組裝3條并行鏈路的結果一起返回
3.2.2 代碼編寫
這里基于上述的梳理出來的圖例寫一下具體的代碼
public CompletableFuture<CFResponse> getResult(){
//并行3條鏈路
CompletableFuture<CF1Response> cf1 = cf1Service.getResult();
CompletableFuture<CF2CombineResponse> cf2Combine = getCf2Combine();
CompletableFuture<CF3CombineResponse> cf3Combine = getCf3Combine();
//組合3個future，轉化結果
CompletableFuture<Void> finalFuture = CompletableFuture.allOf(cf1, cf2Combine, cf3Combine);
return finalFuture.thenApply((unused, r) -> new CFResponse(cf1.get().getCf1Value() +
cf2Combine.get().getCf2CombineValue() + cf3Combine.get().getCf3CombineValue()));
}
//第二條鏈路的執行
private CompletableFuture<CF2CombineResponse> getCf2Combine() {
CompletableFuture<CF2Response> cf2 = cf2Service.getResult();
return cf2.thenCompose(cf2Response -> {
CompletableFuture<CF4Response> cf3 = cf4Service.getResult(cf2Response.getCf2Value());
return cf3.thenApply(cf4Response -> new CF2CombineResponse(cf4Response.getCf4Value()));
});
}
//第三條鏈路的執行
private CompletableFuture<CF3CombineResponse> getCf3Combine() {
CompletableFuture<CF3Response> cf3 = cf3Service.getResult();
return cf3.thenCompose(cf3Response -> {
CompletableFuture<CF5Response> cf5 = cf5Service.getResult(cf3Response.getCf3Value());
CompletableFuture<CF6Response> cf6 = cf6Service.getResult(cf3Response.getCf3Value());
return CompletableFuture.allOf(cf5, cf6).thenCompose(unused -> cf7Service.getResult(cf5.get().getCf5Value(), cf6.get().getCf6Value()));
});
}實際改造代碼片段
接口：
public interface RecommendAtConnectApi {
/**
* 聊天頁
* @param request
* @return
*/
Result<RecommendAtConnectRes> getRecommendContentNew(RecommendAtConnectReq request);
/**
* 聊天頁異步
* @param request
* @return
*/
CompletableFuture<Result<RecommendAtConnectRes>> asyncGetRecommendContentNew(RecommendAtConnectReq request);
}thenApply結果轉化
public CompletableFuture<RecommendAtConnectRes> asyncGetRecommendContent(RecommendAtConnectReq request) {
RecommendAtConnectRequest recommendAtConnectRequest = getRecommendAtConnectRequest(request);
CompletableFuture<RecommendAtConnectDto> future = recommendAtConnectEventHandlerChain.asyncHandlerOfRecommendAtConnect(recommendAtConnectRequest);
return Objects.isNull(future)? null: future.thenApply(this::dtoToRes);
}前后future依賴：
//future編排
CompletableFuture<Optional<RecommendAtConnectDto>> taskEngineFuture = pushGsTaskEngineHandler.asyncPushHandler(connectRequest);
CompletableFuture<Optional<RecommendAtConnectDto>> refundFuture = getNextFuture(taskEngineFuture, connectRequest, unused ->pushLogisticsRefundHandler.asyncPushHandler(connectRequest));
CompletableFuture<Optional<RecommendAtConnectDto>> serviceCaseFuture = getNextFuture(refundFuture, connectRequest, unused ->pushServiceCaseHandler.asyncPushHandler(connectRequest));
CompletableFuture<Optional<RecommendAtConnectDto>> orderFuture = getNextFuture(serviceCaseFuture, connectRequest, unused->pushOrderSourcePredictHandler.asyncPushHandler(connectRequest));
CompletableFuture<Optional<RecommendAtConnectDto>> spuFuture = getNextFuture(orderFuture, connectRequest, unused->pushSpuSourcePredictHandler.asyncPushHandler(connectRequest));
CompletableFuture<Optional<RecommendAtConnectDto>> customerCenterFuture = getNextFuture(spuFuture, connectRequest, unused->pushCustomerCenterSourcePredictHandler.asyncPushHandler(connectRequest));
CompletableFuture<Optional<RecommendAtConnectDto>> guessQuestionFuture = getNextFuture(customerCenterFuture,connectRequest, unused -> pushGuessQuestionHandler.asyncPushHandler(connectRequest));
finalFuture = getNextFuture(guessQuestionFuture, connectRequest, unused -> pushWelcomeHandler.asyncPushHandler(connectRequest));
//回調工具方法
public static CompletableFuture<Optional<RecommendAtConnectDto>> getNextFuture(CompletableFuture<Optional<RecommendAtConnectDto>> beforeFuture,
RecommendAtConnectRequest request,
Function<RecommendAtConnectRequest, CompletableFuture<Optional<RecommendAtConnectDto>>> function) {
return beforeFuture.thenCompose(recommendAtConnectDto -> {
if (!recommendAtConnectDto.isPresent()) {
return function.apply(request);
}
return beforeFuture;
});
}3.2.3 線程池
自定義業務線程池
處理具體的業務邏輯時，如果不傳入線程池，默認使用ForkJoinPool的commonPool，其線程數量默認是CPU的核心數量-1，推薦傳入自定義的業務線程池，防止阻塞dubbo線程。
//自定義dubbo業務線程池
@Bean(name = "dubboAsyncBizExecutor")
public ThreadPoolTaskExecutor dubboAsyncBizExecutor(){
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(200);
executor.setMaxPoolSize(200);
executor.setQueueCapacity(50);
executor.setThreadNamePrefix("dubboAsyncBizExecutor-");
executor.setRejectedExecutionHandler((r, executor1) -> log.error("dubbo async biz task exceed limit"));
return executor;
}
public CompletableFuture<Result<GuessQuestionResponse>> asyncPredictQuestion(PredictQuestionExtRequest request) {
log.info("asyncPredictQuestion start");
CompletableFuture<Result<GuessQuestionResponse>> resultCompletableFuture =
CompletableFuture.supplyAsync(() -> predictQuestionNew(request), dubboAsyncBizExecutor);
log.info("asyncPredictQuestion end");
return resultCompletableFuture;
}同步和異步線程隔離（目前最新正式版本3.2.0支持）
<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:dubbo="http://dubbo.apache.org/schema/dubbo"
xmlns="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.3.xsd
http://dubbo.apache.org/schema/dubbo http://dubbo.apache.org/schema/dubbo/dubbo.xsd">
<!-- NOTE: we need config executor-management-mode="isolation" -->
<dubbo:application name="demo-provider" executor-management-mode="isolation">
</dubbo:application>
<bean id="syncService" class="org.apache.dubbo.config.spring.impl.SyncServiceImpl"/>
<bean id="asyncService" class="org.apache.dubbo.config.spring.impl.AsyncServiceImpl"/>
<!-- customized thread pool -->
<bean id="executor-sync-service"
class="org.apache.dubbo.config.spring.isolation.spring.support.SyncServiceExecutor"/>
<bean id="executor-async-service"
class="org.apache.dubbo.config.spring.isolation.spring.support.AsyncServiceExecutor"/>
<dubbo:service executor="executor-sync-service"
interface="org.apache.dubbo.config.spring.api.SyncService" version="1.0.0"
timeout="3000" ref="syncService" />
<dubbo:service executor="executor-async-service"
interface="org.apache.dubbo.config.spring.api.AsyncService" version="1.0.0"
timeout="5000" ref="asyncService" />
</beans>3.2.4 異常處理
CompletableFuture異常處理使用回調exceptionally，當CompletableFuture執行的過程拋出了異常，會使用CompletionException進行封裝然后拋出。
CompletableFuture<RecommendAtConnectDto> asyncPushContent(RecommendAtConnectRequest connectRequest) {
//業務方法，內部會發起異步rpc調用
CompletableFuture<String> future = orderSourcePredictHandlerChain.asyncHandleOfPredict(connectRequest);
//這里回調方法thenApply，如果發生異常thenApply內部會通過new CompletionException(throwable) 對異常進行包裝
return Objects.isNull(future)? null : future.thenApply(messageBody->{
if (StrUtil.isBlank(messageBody)){
log.info(" async orderSourcePredictHandlerChain.handleOfPredict fail, connectRequest:{}", JSON.toJSONString(connectRequest));
return null;
}
RecommendAtConnectDto connectDto = RecommendAtConnectDtoUtil.getDto(
messageBody, connectRequest.getSessionId(),
connectRequest.getCreateChatReq().getUserId(), MessageBodyTypeEnum.MULTI_STAGE.getCode(), EventEnum.PUSH_MULTI_STAGE_MESSAGE.getCode());
return connectDto;
}).exceptionally(err -> {
//通過exceptionally 捕獲異常，這里的err已經被thenApply包裝過，因此需要通過Throwable.getCause()提取異常
log.error("orderSourcePredictHandlerChain.handleOfPredict Exception cnotallow={}", JSON.toJSONString(connectRequest), ExceptionUtils.extractRealException(err));
return 0;
});
}異常使用自定義工具類ExceptionUtils進行提取。
public class ExceptionUtils {
public static Throwable extractRealException(Throwable throwable) {
//這里判斷異常類型是否為CompletionException、ExecutionException，如果是則進行提取，否則直接返回。
if (throwable instanceof CompletionException || throwable instanceof ExecutionException) {
if (throwable.getCause() != null) {
return throwable.getCause();
}
}
return throwable;
}
}3.2.5 穩定性保障
- 改造的過程從上到下改動的同步方法保持不變，新增異步的方法進行支持
- 改造的接口是上游服務端依賴的，和上游服務端溝通，通過AB控制調用同步和異步接口
- 改造的接口是App端依賴的，在接口實現處通過AB控制調用異步和同步service
- 通過以上三種方法可以實現一鍵回滾到最初的邏輯
3.3 遇到的問題
- CompletableFuture回調方法中打印的日志會丟失traceId，已找監控團隊幫忙支持解決，但是會增加應用gc的次數，現在生產上是白名單應用開放中
- 異步接口線程池和同步接口線程池隔離在dubbo最新發布的正式版本3.2.0支持
- CompletableFuture.thenCompose不支持返回null，需要將返回值用Optional包裝返回
- 打印日志的位置變更，由于返回值是future，拿不到真實的結果，只能在回調之中打印日志才能看到真實的結果
- 監控平臺監控的平均耗時不包含回調的耗時，對于排查接口性能問題會增加一些難度，例如5月10日遇到了一個異步接口耗時同比增加了50%，但是從監控平臺上看到平均耗時并沒有明顯增加
4、異步化收益
- 壓測接口性能提升了50%
- 線上接口RT降低1/4左右，其中輸入聯想接口RT由173.04ms降為119.43ms
- 服務器資源縮減了1/3
- 服務器資源利用率提升
異步化之前CPU的使用率：
異步化縮減機器之后CPU的使用率：
可以看到dubbo異步化之后，服務器cpu的使用率由18左右提升到了50%左右，大家在進行機器縮減時需要關注一下CPU的使用率，當CPU的使用率超過60%時就會引發報警，這個就是我們縮減的極限了，如果在繼續縮減在一些流量高峰或者流量飆升的場景會出現風險。
5、其他
- Dubbo異步化對編程者的代碼水平和架構能力都有一定的要求，同時在對老的代碼異步化的過程中，通過對上述接口先后調用關系的梳理也能發現很多代碼不合理或者有性能問題的地方，對代碼質量的提高也有一定的好處，其實就算不是想異步化，而是想提高代碼的并發度，這種前后依賴關系的梳理也是必不可少的，只不過異步化是將程序的并發度提升到極致的一種表現。
- Dubbo異步化編程和以往的同步編程習慣可能有所不同，但是轉念一想，是不是異步化才是現實世界中更加真實的寫照，更加的符合現實世界運轉的規律，我們在規劃做一件事情時，往往會將事情進行拆解，然后同時（是指同一段時間不是同一刻）去做沒有先后依賴關系的多件事情，而不是做一件事，然后一直等到有結果了再去做其他事情。
- 通過壓測我們發現當壓測qps不斷提高依賴的接口或者組件的耗時增加比較明顯，且慢慢成為性能提升的瓶頸時，異步帶來的提升效果會受到此瓶頸的制約，帶來提升會有一定比例的折扣，所以大家在做異步化實踐時，需要稍微降低一些提升的預期。
6、總結
通過這次實踐，我們使用CompletableFuture將Dubbo接口進行了異步化，同時利用CompletableFuture的異步回調能力，減少了服務依賴之間的阻塞，增加了dubbo線程的處理請求的能力，同時利用CompletableFuture傳入的業務線程提高了服務器CPU資源的利用率，用更少的硬件資源可以處理更多的請求，為公司的降本增效貢獻了一小份力量。
責任編輯：武曉燕
來源：
得物技術
