精选｜Dubbo异步化实践

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 actions

CompletableFuture有两个非常重要的属性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 connectRequest={}", 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资源的利用率，用更少的硬件资源可以处理更多的请求，为公司的降本增效贡献了一小份力量。

线下活动推荐

时间：2023年6月10日（周六） 14:00-18:00

主题：得物技术沙龙总第18期-无线技术第4期

地点：杭州·西湖区学院路77号得物杭州研发中心12楼培训教室（地铁10号线&19号线文三路站G口出）

活动亮点：本次无线沙龙聚焦于最新的技术趋势和实践，将在杭州/线上为你带来四个令人期待的演讲话题，包括：《抖音创作工具-iOS功耗监控与优化》、《得物隐私合规平台建设实践》、《网易云音乐-客户端大流量活动的日常化保障方案实践》、《得物Android编译优化》。相信这些话题将对你的工作和学习有所帮助，我们期待着与你共同探讨这些令人兴奋的技术内容！

报名方式：点击报名无线技术沙龙

在这里插入图片描述