我删掉了自己写的 300 行 Future，把整个 RPC 框架改成全链异步

Jaws 系列第 6 篇。前情提要《删掉 gRPC 依赖后我用 2400 行打通了 gRPC 生态》、《HTTP/2 传输进化史》。本文代码全部出自 javahongxi/jawscommit 可查。一、一个让人不舒服的 join()故事要从 wire 模块的一次重构说起。8 月底我给 jaws-wire 加非阻塞分发时WireCallDispatcher里有一段代码让我越看越难受CompletableFutureObjectfuturemessageHandler.handleAsync(jawsRequest);// 旧版Responseresultfuture.join();// 业务线程就这么被占着等handleDispatchResult(result,ctx,serverHandler);这是 gRPC 线格式的 Provider 管线分发路径。handleAsync明明返回的是CompletableFuture——异步语义都到位了——下游却一个join()把结果等回来。业务方法快还好一旦下游是慢调用每个在途请求都占着一个业务线程干等。线程数就是并发的天花板这等于把异步框架用出了同步的损耗。改成什么样先把答案放这里CompletableFutureObjectfuturemessageHandler.handleAsync(jawsRequest);if(future.isDone()){// 快路径业务方法同步返回时直接内联处理不切线程handleDispatchResult(future.join(),ctx,serverHandler);}else{// 慢路径挂回调业务线程立刻释放future.whenComplete((result,throwable)-{handleDispatchResult(result,ctx,serverHandler);});}isDone()快路径是个值得停下来品一下的细节大多数业务方法其实跑得飞快future 返回时已经完成了这时候再挂回调纯属浪费一次线程切换。所以先探一下已完成就内联join()此时 join 不阻塞没完成才走whenComplete。一行判断换掉一次无谓的调度开销。改完 wire 这段我以为这事就结束了。直到我顺着调用链往回看了一眼客户端——才发现真正的题目有多大。二、顺着调用链往上爬RPC 框架的调用链从消费端到服务端大致是这样一条河业务代码 → 代理层 → Filter 链 → Cluster/LB → Reference → 序列化 → 传输层 ↓ (网络) 传输层 ← 序列化 ← Provider 管线 ← Filter 链 ← 业务实现wire 的join()只是河下游的一个洞。我从洞往上游走一路上看到的是这样的景象Filter 契约是同步的。jaws 的Filter接口长这样publicinterfaceFilter{Responsefilter(Caller?caller,Requestrequest);}返回Response意味着什么意味着 Filter 必须拿到最终结果才能返回。而 Filter 的下一跳caller.call(request)走到传输层传输层得等网络回包——于是每个 Filter 都被迫阻塞等一次完整的 RPC。更别提链式组装了三个 Filter 串起来就是三次「等到天荒地老」的串行阻塞。TracingFilter 已经在偷偷变形。最典型的是链路追踪 Filter改造前的消费端逻辑// 改造前同步契约下的无奈写法privateResponsehandleConsumer(...){Spanspant.nextSpan().name(spanName).start();try(Tracer.SpanInScopescopet.withSpan(span)){p.inject(span.context(),request,Request::setAttachment);Responseresponsecaller.call(request);// 阻塞等整个 RPCif(response.getException()!null){span.error(response.getException());}returnresponse;}catch(Exceptione){span.error(e);throwe;}finally{span.end();}}这段代码有个隐藏的问题span.end()在 finally 里时机是「RPC 结束」——没错但前提是线程一直停在这里等。span 的生命周期和线程的阻塞生命周期被迫绑死了。想做到「请求发出后线程就走span 等回包时再关」在同步契约下根本写不出来。DefaultResponseFuture 是一只自研怪兽。客户端等回包的核心类改造前 252 行自带状态机publicclassDefaultResponseFutureimplementsResponseFuture{protectedvolatileFutureStatestateFutureState.DOING;// 自研状态枚举protectedvolatileObjectresult;protectedvolatileExceptionexception;protectedvolatileListFutureListenerlisteners;// 自研监听器publicObjectgetValue(){synchronized(this){if(!isDoing()){returngetValueOrThrow();}longwaitTimetimeout-(System.currentTimeMillis()-createTime);if(waitTime0){for(;;){try{wait(waitTime);// 手写 wait/notify}catch(InterruptedExceptionignore){Thread.currentThread().interrupt();}if(!isDoing())break;// 重新计算剩余超时继续 wait …}}if(isDoing())cancelOnTimeout();returngetValueOrThrow();}}}手写 wait/notify 循环、手写超时补偿、手写 listener 通知、FutureState三态枚举、FutureListener回调接口……这些 JDK 在CompletableFuture里全部都有而且经过十几年生产环境锤炼。我维护这只怪兽的每一行都在重新发明 JDK 已经解决的问题。服务端管线也在同步裸奔。顺着河再看服务端一侧MessageHandler是传输层和业务之间的桥它在旧版里的签名是同步的Response handle(Object message)。Netty 的NettyChannelHandler收到请求后得等handle返回才能把响应写回 channel——事件循环线程event loop被业务调用整段占住。而事件循环是 Netty 的命根子一个 event loop 管着几十条连接它被卡 100ms这几十条连接上的所有请求全部顺延。这不是「慢一点」的问题是所有连接互相拖累的放大器。到这里问题已经从「wire 有个 join 不顺眼」升级成了框架名字叫 JAWSJava Async Wire Service异步却只做到了传输层往上一点点——Filter 拦在中间Future 是同步内核整条链是异步的躯干拖着同步的四肢。三、决策契约先行从 Filter 开刀改造顺序是这次最有意思的决策点。可选路径有三条先改传输层把 Netty/http2 内部全异步化Filter 契约不动——治标join 只是藏得更深先改 Filter 契约filter()返回CompletableFutureResponse逼着整条链跟着改先改 FutureDefaultResponseFuture 换成 CompletableFuture上层不动我选了 2契约先行。理由契约是骨架骨架定了肉才知道往哪长。Filter.filter()的返回类型一变编译器会把所有「不改就会坏」的地方精确地列出来——Filter 实现、包装器、调用方一个都逃不掉。这比人肉排查安全得多。新的Filter接口SpipublicinterfaceFilter{CompletableFutureResponsefilter(Caller?caller,Requestrequest);}配套在Caller jaws 里消费端 Reference 和服务端 Provider 的共同抽象对标 Dubbo 的 Invoker上加默认方法publicinterfaceCallerTextendsEndpoint{Responsecall(Requestrequest);defaultCompletableFutureResponsecallAsync(Requestrequest){returnCompletableFuture.completedFuture(call(request));}}callAsync()做成 default 方法是个关键的兼容设计新契约是「邀请」而不是「强拆」。老实现不改照样编译通过默认桥接到同步 call新的实现可以按自己的节奏迁到异步。整个迁移过程中测试始终是绿的。TracingFilter 的蜕变契约一换之前写不出来的代码自然就长出来了// 改造后span 生命周期不再绑死线程privateCompletableFutureResponsehandleConsumer(...){Spanspant.nextSpan().name(spanName).start();try(Tracer.SpanInScopescopet.withSpan(span)){p.inject(span.context(),request,Request::setAttachment);}returncaller.callAsync(request).whenComplete((response,throwable)-{if(throwable!null){span.error(throwable);}elseif(response!nullresponse.getThrowable()!null){span.error(response.getThrowable());}span.end();// 回包时关 span线程早就走了});}对比一下两版的本质差异注入 trace 上下文这种「出发前」的动作在调用线程完成span.end()挪进whenComplete变成「到达后」的动作由完成 future 的那个线程执行。span 的生命周期终于和 RPC 的生命周期对齐而不是和某个线程的阻塞周期对齐。这就是异步契约的价值——让代码的形状贴住事物的真实形状。AccessLog、TokenAuth、Metrics 四个内置 Filter 全部照此迁移全部改成callAsync().whenComplete()的非阻塞后处理模式。FilterProviderWrapper一个 join 也不留Filter 链的组装节点是FilterProviderWrapper它的双轨实现最能说明这次升级的彻底性OverridepublicResponsecall(Requestrequest){if(isFilterDisabled(request.getInterfaceName())){returnoriginal.call(request);}returnfilter.filter(original,request).join();// 同步门面内部一次 join}OverridepublicCompletableFutureResponsecallAsync(Requestrequest){if(isFilterDisabled(request.getInterfaceName())){returnoriginal.callAsync(request);}returnfilter.filter(original,request);// 原生异步零 join}同步入口call()保留给确实需要阻塞语义的调用方内部一次 join 封装异步入口callAsync()则全程无 join 直通。框架内部全部走callAsync同步门面只留给用户边界。四、把 252 行怪兽换成一行 extendsFilter 契约改完轮到那只怪兽了。DefaultResponseFuture的改造方向其实早在选型时就定了JDK 的CompletableFuture已经把状态机、监听器、超时、组合算子全部做好自研的意义只剩「历史包袱」。于是改造后的全部声明是——publicclassDefaultResponseFutureextendsCompletableFutureResponseimplementsResponseFuture{privatefinalRequestrequest;privatefinalinttimeout;OverridepublicvoidonSuccess(Responseresponse){complete(response);}OverridepublicvoidonFailure(Responseresponse){completeExceptionally(response.getThrowable());}}252 行瘦身到 128 行——剩余部分基本是 Javadoc 和给老 API 用的异常转换。删掉的东西列一下全是 JDK 替我保管的FutureState三态枚举 →CompletableFuture内部状态机FutureListener 手写通知循环 →whenComplete/thenApply手写 wait/notify 超时循环 → 传输层每请求定时器 completeExceptionally手写的getRawValue()/getThrowable()状态判读 →getNow(null)/isCompletedExceptionally()组合保留ResponseFuture门面接口是为了 API 稳定getValue()/getTimeout()/getRequestId()这些老签名还在但内核已经完全是 CompletableFuture 了。阻塞式用户代码getValue()委托给get()异步用户代码直接whenComplete同一只 future 两种活法。有个容易被忽略的收益藏在 commit message 里「消除双 Future 分配开销和 monitor lock 成本」。旧实现里 Reference 层为了拿到 CompletableFuture 语义要再 new 一个 CompletableFuture 把 ResponseFuture 桥接过去——每个请求两只 future、一次锁。现在DefaultResponseFuture本身就是 CompletableFutureAbstractReference 里直接whenComplete链上去一只 future 贯穿到底// AbstractReference.callAsync网络层完成 future 时顺路做统计if(responseinstanceofCompletableFuture?cf){returncf.whenComplete((r,t)-{decrActiveCount(response);if(tnull){longelapsedSystem.nanoTime()-startTime;succeededElapsed.addAndGet(elapsed);succeededCount.incrementAndGet();}}).thenApply(r-(Response)r);}调用统计activeCount、成功耗时累计本来要靠额外回调挂钩现在就是 future 链上的一环。基础设施统一之后横切逻辑的挂载方式自然就变优雅了——这是我认为比「少 300 行」更值钱的部分。五、收尾语义对齐与三个细节主战役之外这次还顺手做了三件值得单独一说的事。异常语义对齐。getException全链路改名getThrowable类型从Exception放宽到Throwable。别小看这个放宽序列化栈溢出抛的是StackOverflowError同步业务代码抛Error子类的场景也不罕见老契约里这些直接被类型系统拒收只能丢信息。改名加放宽是一起还的旧债。服务端管线 handleAsync。前面埋的伏笔在这里兑现MessageHandler的签名从同步handle()换成handleAsync()返回CompletableFutureObject由AbstractRequestHandler统一实现——Provider 查找、方法解析这些前置逻辑照旧在事件循环完成纯内存操作快且安全真正的业务调用交给doHandleAsync。NettyChannelHandler收到请求后改成挂whenComplete回调回包到达时再把响应写回 channel。事件循环线程从此只做「接请求、写响应」两件事中间漫长的业务执行与它无关。熔断计数修正。WireClient 的错误熔断要对齐 Http2Client 的whenComplete语义——区分「业务异常」和「框架错误」业务抛JawsBizException说明服务本身活着只是这次调用失败不计入熔断网络错误、超时这类框架级错误才incrErrorCount()成功后resetErrorCount()。区分的意义在于避免业务侧正常的异常流量把熔断器误打开——框架挂了才该熔断业务出错不该。用户侧异步 API 落地。契约改完了得让用户用得上。DemoService接口补了helloAsyncString 返回和getUserAsyncPOJO 返回两个异步方法NettyConsumer 里演示了从基础回调到并发组合的完整用法// 基础异步提交后线程立刻返回回调里拿结果CompletableFutureStringasyncHellodemoService.helloAsync(async-lily);asyncHello.thenAccept(result-System.out.println(callback result));// thenApply 链式变换demoService.helloAsync(chain-demo).thenApply(String::toUpperCase).thenAccept(s-System.out.println(chained s));// allOf 并发组合两个调用齐了再汇总CompletableFutureStringf1demoService.helloAsync(user-A);CompletableFutureStringf2demoService.helloAsync(user-B);CompletableFuture.allOf(f1,f2).thenRun(()-System.out.println(combined [f1.join(), f2.join()]));六、W 落定Async Wire Service 名实对齐回头看这次改造的完整地图层改造前改造后Filter 契约Response filter(...)同步拦截CompletableFutureResponse filter(...)异步管道Caller 抽象只有call()新增callAsync()default 桥接客户端 Future252 行自研 wait/notify 状态机extends CompletableFuture128 行传输层分发future.join()占线程isDone()快路径 whenComplete回调服务端管线同步handle()handleAsync()返回 CompletableFuture流式wire/http2 各自实现共享StreamPublisherFlow.Publisher 基座JAWS 这个名字立项目时就起好了——JavaAsyncWireService。但说实话改造之前那个 A 是有水分的异步只在传输层往上一点点Filter 拦腰一截同步客户端内核是 wait/notify。这次全链贯通之后从业务代理、Filter、Cluster、Reference 到传输层再到 wire 分发CompletableFuture 一竿子插到底A 字才算名副其实。复盘这次改造有三条经验值得留给读者契约先于实现。改返回类型这种「编译器帮你找全改动点」的路径永远比人肉排查安全。同步入口保留为门面、异步作为原生路径的双轨设计让迁移过程没有一天是红的。删自研代码之前先确认标准库真的覆盖了你的场景。CompletableFuture 覆盖了状态机、监听、超时、组合而 jaws 特有的每请求超时归传输层定时器管理——边界划清了删起来才不心虚。顺着调用链找下一个洞。wire 的一个 join() 只是症状从症状出发往上游走完整条链才能看到问题的全貌。单点优化容易全链对齐难难就难在要舍得把「已经能跑」的代码推翻。jaws 现在核心四模块约 2.3 万行README 里那句「可以从头读到尾」依然成立——而且这次重构之后「读」的体验更好了你顺着一条callAsync从 Filter 走到传输层看到的将是同一种异步语言而不是三种方言的翻译现场。下一篇候选是自适应负载均衡的实现剖析power of two choices 在 RPC 里的落地感兴趣的可以先去仓库读AdaptiveLoadBalance我们下篇见。项目地址github.com/javahongxi/jaws — 核心约 2.3 万行、可从头读到尾的轻量级 RPC 框架。三传输Netty 二进制 / HTTP/2 / gRPC 线格式gRPC 互通Server Streaming自适应负载均衡实测 10 万 QPS。