Dubbo异步化实践
1、背景
从Apach Dubbo的官网了解到从 2.7.0 版本开始，Dubbo 的所有异步编程接口开始以CompletableFuture为基础，Dubbo接口异步化能够极大地提高接口性能，降低接口依赖调用之间的阻塞，同时了解到我们公司大部分应用使用的是同步rpc，在公司降本增效的大背景下，我们选择了在客服机器人组对Dubbo异步化进行落地实践，实践下来发现Dubbo异步化对接口性能提升了50%，涉及异步化的应用服务器缩减了1/3，接下来主要为大家分享一下实践的经验以及异步化提升的效果。
(资料图片)
通过CompletableFuture可以将复杂的业务逻辑从Dubbo线程池（大小默认200）切换到用户自定义的业务线程来执行，提升Dubbo线程池请求的处理能力，同时增加自定义业务线程池，提升服务器的资源利用率。接下来我们来看下CompletableFuture怎么异步化Dubbo接口以及其原理。
2.1 接口改造方式getRecommendContent为老的方法，asyncGetRecommendContent为新添加的异步方法；老的方法保留，兼容没有升级的调用方；添加新的异步方法，返回值使用CompletableFuture进行包装：
public interface RecommendAtConnectApi { Result getRecommendContent(RecommendAtConnectReq request); CompletableFuture> asyncGetRecommendContent(RecommendAtConnectReq request);} 下面先介绍几种常用的使用方式：
future的结果获取到时转化处理（thenApply）CompletableFuture cFuture = cAsyncService.asyncSayHello(name);CompletableFuture finalFuture = cFuture.thenApply(c -> new DataDTO());return finalFuture; CompletableFuture cFuture = cAsyncService.asyncSayHello(name);CompletableFuture dFuture = dAsyncService.asyncSayHello(name);CompletableFuture allFuture = cFuture.thenCombine(dFuture, (c, d) -> new DataDTO());return allFuture; CompletableFuture> taskEngineFuture = pushGsTaskEngineHandler.asyncPushHandler(connectRequest);CompletableFuture> refundFuture = getNextFuture(taskEngineFuture, connectRequest, unused ->pushLogisticsRefundHandler.asyncPushHandler(connectRequest));return refundFuture;//回调工具方法public static CompletableFuture> getNextFuture(CompletableFuture> beforeFuture, RecommendAtConnectRequest request, Function>> 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 AltResultvolatile Completion stack; // Top of Treiber stack of dependent actionsCompletableFuture有两个非常重要的属性result和stack，result是future中的结果，stack是future获取到结果时回调的函数动作存储的栈，stack是一个Completion，Completion中有指向下一个Completion的指针。
thenApply//thenApply源码private CompletableFuture uniApplyStage( Executor e, Function f) { if (f == null) throw new NullPointerException(); CompletableFuture d = new CompletableFuture(); if (e != null || !d.uniApply(this, f, null)) { //生成当前future的依赖 UniApply c = new UniApply(e, d, this, f); //CAS操作压入栈中 push(c); //尝试运行一下 c.tryFire(SYNC); } return d;} thenApply的原理比较简单在调用的时候会将回调的逻辑生成UniApply压入栈中，UniApply中包含了返回的future和当前的feture，等到当前future有结果返回时，会回调执行栈中的函数f。
thenCombine//thenCombine源码private CompletableFuture biApplyStage( Executor e, CompletionStage o, BiFunction f) { CompletableFuture b; if (f == null || (b = o.toCompletableFuture()) == null) throw new NullPointerException(); CompletableFuture d = new CompletableFuture(); if (e != null || !d.biApply(this, b, f, null)) { //生成二元依赖的BiCompletion BiApply c = new BiApply(e, d, this, b, f); //将其压入当前和组合的future栈中 bipush(b, c); c.tryFire(SYNC); } return d;} thenCombine依赖两个future，返回一个新的future，当依赖的两个future都有结果返回之后，回调传入的函数动作。
thenCompose//thenCompose源码private CompletableFuture uniComposeStage( Executor e, Function> 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(encodeThrowable(x, r)); } r = null; } try { @SuppressWarnings("unchecked") T t = (T) r; //将当前处理结果作为f的输入，并执行f得到新的future g CompletableFuture g = f.apply(t).toCompletableFuture(); Object s = g.result; //如果已经有结果直接返回 if (s != null) return new CompletableFuture(encodeRelay(s)); //new一个返回的future CompletableFuture d = new CompletableFuture(); //生成一个元依赖的UniCompletion UniRelay copy = new UniRelay(d, g); //将其压入g的栈中 g.push(copy); copy.tryFire(SYNC); return d; } catch (Throwable ex) { return new CompletableFuture(encodeThrowable(ex)); } } //如果当前结果为空，则直接生成当前feture的依赖，压入栈中 CompletableFuture d = new CompletableFuture(); UniCompose c = new UniCompose(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返回的CompletableFutureCF1、CF2和CF3同一层的代表它们是并行的关系，CF2和CF4前后代表它们是依赖的关系最后组装3条并行链路的结果一起返回3.2.2 代码编写这里基于上述的梳理出来的图例写一下具体的代码
public CompletableFuture getResult(){ //并行3条链路 CompletableFuture cf1 = cf1Service.getResult(); CompletableFuture cf2Combine = getCf2Combine(); CompletableFuture cf3Combine = getCf3Combine(); //组合3个future，转化结果 CompletableFuture finalFuture = CompletableFuture.allOf(cf1, cf2Combine, cf3Combine); return finalFuture.thenApply((unused, r) -> new CFResponse(cf1.get().getCf1Value() + cf2Combine.get().getCf2CombineValue() + cf3Combine.get().getCf3CombineValue()));}//第二条链路的执行private CompletableFuture getCf2Combine() { CompletableFuture cf2 = cf2Service.getResult(); return cf2.thenCompose(cf2Response -> { CompletableFuture cf3 = cf4Service.getResult(cf2Response.getCf2Value()); return cf3.thenApply(cf4Response -> new CF2CombineResponse(cf4Response.getCf4Value())); });}//第三条链路的执行private CompletableFuture getCf3Combine() { CompletableFuture cf3 = cf3Service.getResult(); return cf3.thenCompose(cf3Response -> { CompletableFuture cf5 = cf5Service.getResult(cf3Response.getCf3Value()); CompletableFuture 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 getRecommendContentNew(RecommendAtConnectReq request); /** * 聊天页异步 * @param request * @return */ CompletableFuture> asyncGetRecommendContentNew(RecommendAtConnectReq request);} public CompletableFuture asyncGetRecommendContent(RecommendAtConnectReq request) { RecommendAtConnectRequest recommendAtConnectRequest = getRecommendAtConnectRequest(request); CompletableFuture future = recommendAtConnectEventHandlerChain.asyncHandlerOfRecommendAtConnect(recommendAtConnectRequest); return Objects.isNull(future)? null: future.thenApply(this::dtoToRes);} 前后future依赖：
//future编排CompletableFuture> taskEngineFuture = pushGsTaskEngineHandler.asyncPushHandler(connectRequest);CompletableFuture> refundFuture = getNextFuture(taskEngineFuture, connectRequest, unused ->pushLogisticsRefundHandler.asyncPushHandler(connectRequest));CompletableFuture> serviceCaseFuture = getNextFuture(refundFuture, connectRequest, unused ->pushServiceCaseHandler.asyncPushHandler(connectRequest));CompletableFuture> orderFuture = getNextFuture(serviceCaseFuture, connectRequest, unused->pushOrderSourcePredictHandler.asyncPushHandler(connectRequest));CompletableFuture> spuFuture = getNextFuture(orderFuture, connectRequest, unused->pushSpuSourcePredictHandler.asyncPushHandler(connectRequest));CompletableFuture> customerCenterFuture = getNextFuture(spuFuture, connectRequest, unused->pushCustomerCenterSourcePredictHandler.asyncPushHandler(connectRequest));CompletableFuture> guessQuestionFuture = getNextFuture(customerCenterFuture,connectRequest, unused -> pushGuessQuestionHandler.asyncPushHandler(connectRequest));finalFuture = getNextFuture(guessQuestionFuture, connectRequest, unused -> pushWelcomeHandler.asyncPushHandler(connectRequest));//回调工具方法public static CompletableFuture> getNextFuture(CompletableFuture> beforeFuture, RecommendAtConnectRequest request, Function>> function) { return beforeFuture.thenCompose(recommendAtConnectDto -> { if (!recommendAtConnectDto.isPresent()) { return function.apply(request); } return beforeFuture; });} 处理具体的业务逻辑时，如果不传入线程池，默认使用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> asyncPredictQuestion(PredictQuestionExtRequest request) { log.info("asyncPredictQuestion start"); CompletableFuture> resultCompletableFuture = CompletableFuture.supplyAsync(() -> predictQuestionNew(request), dubboAsyncBizExecutor); log.info("asyncPredictQuestion end"); return resultCompletableFuture;} CompletableFuture异常处理使用回调exceptionally，当CompletableFuture执行的过程抛出了异常，会使用CompletionException进行封装然后抛出。
CompletableFuture asyncPushContent(RecommendAtConnectRequest connectRequest) { //业务方法，内部会发起异步rpc调用 CompletableFuture 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; }}异步化之前CPU的使用率：
异步化缩减机器之后CPU的使用率：
可以看到dubbo异步化之后，服务器cpu的使用率由18左右提升到了50%左右，大家在进行机器缩减时需要关注一下CPU的使用率，当CPU的使用率超过60%时就会引发报警，这个就是我们缩减的极限了，如果在继续缩减在一些流量高峰或者流量飙升的场景会出现风险。
5、其他Dubbo异步化对编程者的代码水平和架构能力都有一定的要求，同时在对老的代码异步化的过程中，通过对上述接口先后调用关系的梳理也能发现很多代码不合理或者有性能问题的地方，对代码质量的提高也有一定的好处，其实就算不是想异步化，而是想提高代码的并发度，这种前后依赖关系的梳理也是必不可少的，只不过异步化是将程序的并发度提升到极致的一种表现。Dubbo异步化编程和以往的同步编程习惯可能有所不同，但是转念一想，是不是异步化才是现实世界中更加真实的写照，更加的符合现实世界运转的规律，我们在规划做一件事情时，往往会将事情进行拆解，然后同时（是指同一段时间不是同一刻）去做没有先后依赖关系的多件事情，而不是做一件事，然后一直等到有结果了再去做其他事情。通过压测我们发现当压测qps不断提高依赖的接口或者组件的耗时增加比较明显，且慢慢成为性能提升的瓶颈时，异步带来的提升效果会受到此瓶颈的制约，带来提升会有一定比例的折扣，所以大家在做异步化实践时，需要稍微降低一些提升的预期。6、总结通过这次实践，我们使用CompletableFuture将Dubbo接口进行了异步化，同时利用CompletableFuture的异步回调能力，减少了服务依赖之间的阻塞，增加了dubbo线程的处理请求的能力，同时利用CompletableFuture传入的业务线程提高了服务器CPU资源的利用率，用更少的硬件资源可以处理更多的请求，为公司的降本增效贡献了一小份力量。
X 关闭
