當(dāng)前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

Dubbo 3.0 预览版解读，6到飞起~

發(fā)布時(shí)間：2025/3/21 编程问答 22 豆豆

生活随笔收集整理的這篇文章主要介紹了 Dubbo 3.0 预览版解读，6到飞起~ 小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.

來源：阿里巴巴中間件公眾號(hào)

Dubbo 自 2011 年 10 月 27 日開源后，已被許多非阿里系的公司使用，其中既有當(dāng)當(dāng)網(wǎng)、網(wǎng)易考拉等互聯(lián)網(wǎng)公司，也不乏中國(guó)人壽、青島海爾等大型傳統(tǒng)企業(yè)。

自去年 12 月開始，Dubbo 3.0?便已正式進(jìn)入開發(fā)階段《重大利好，Dubbo 3.0要來了》，并備受社區(qū)和廣大 Dubbo 用戶的關(guān)注，本文將為您詳細(xì)解讀?3.0 預(yù)覽版的新特性和新功能。

下面先解答一下兩個(gè)有意思的與 Dubbo 相關(guān)的疑問。

為什么 Dubbo 一開源就是 2.0 版本？之前是否存在 1.0 版本？

筆者曾做過 Dubbo 協(xié)議的適配兼容，Dubbo 確實(shí)存在過 1.x 版本，而且從協(xié)議設(shè)計(jì)和模型設(shè)計(jì)上都與 2.0 的開源版本協(xié)議是完全不一樣的。下圖是關(guān)于 Dubbo 的發(fā)展路徑：

阿里內(nèi)部正在使用 Dubbo 開源版本嗎？

是的，非常確定，當(dāng)前開源版本的 Dubbo 在阿里巴巴被廣泛使用，而阿里的電商核心部門是用的 HSF2.2 版本，這個(gè)版本是兼容了 Dubbo 使用方式和 Remoting 協(xié)議。當(dāng)然，我們現(xiàn)在正在做 HSF2.2 的升級(jí)，直接依賴開源版本的 Dubbo 來做內(nèi)核的統(tǒng)一。所以，Dubbo 是得到大規(guī)模線上系統(tǒng)驗(yàn)證的分布式服務(wù)框架，這一點(diǎn)毋容置疑。

Dubbo 3.0 預(yù)覽版的要點(diǎn)

Dubbo 3.0 在設(shè)計(jì)和功能上的新增支持和改進(jìn)，主要是以下四方面：

Dubbo 內(nèi)核之 Filter 鏈的異步化

這里要指出的是，3.0 中規(guī)劃的異步去阻塞和 2.7 中提供的異步是兩個(gè)層面的特性。2.7 中的異步是建立在傳統(tǒng) RPC 中 request – response 會(huì)話模型上的，而 3.0 中的異步將會(huì)從通訊協(xié)議層面由下向上構(gòu)建，關(guān)注的是跨進(jìn)程、全鏈路的異步問題。通過底層協(xié)議開始支持 streaming 方式，不單單可以支持多種會(huì)話模型，還可以在協(xié)議層面開始支持反壓、限流等特性，使得整個(gè)分布式體系更具有彈性。綜上所述，2.7 關(guān)注的異步更局限在點(diǎn)對(duì)點(diǎn)的異步（一個(gè) consumer 調(diào)用一個(gè) provider），3.0 關(guān)注的異步化，寬度上則關(guān)注整個(gè)調(diào)用鏈上的異步，高度上則向上又可以包裝成 Rx 的編程模型。有趣的是，Spring 5.0 發(fā)布了對(duì) Flux 的支持，隨后開始解決跨進(jìn)程的異步問題。
功能方面是 reactive（響應(yīng)式）支持

最近幾年，?reactive programming這個(gè)詞語的熱度迅速提升，Wikipedia 上的 reactive programming 解釋是 reactive programming is a programming paradigm oriented around data flows and the propagation of change. Dubbo3.0會(huì)實(shí)現(xiàn)Reactive Stream 的 rx 接口，從而能讓用戶享受到RP帶來的響應(yīng)性提升，甚至面向 RP 的架構(gòu)升級(jí)。當(dāng)然，我們希望 reactive 不單單能夠帶來事件（event）驅(qū)動(dòng)的應(yīng)用集成方式的升級(jí)，也希望在 Load Balance（選擇最優(yōu)的服務(wù)節(jié)點(diǎn)），fault tolerance（限流降級(jí)時(shí)最好做到自適應(yīng)）等方面發(fā)揮其積極價(jià)值。
云原生/ ServiceMesh 方向的探索

我們定下的策略是進(jìn)入 Envoy 社區(qū)來實(shí)現(xiàn) Dubbo 融入 mesh 的理念思想，目前 Dubbo 協(xié)議已經(jīng)被 Envoy 支持。當(dāng)然，Dubbo Mesh 離真正可用還有很長(zhǎng)一段距離，其在選址、負(fù)載均衡和服務(wù)治理方面的工作需要繼續(xù)在數(shù)據(jù)面建設(shè)，另外，控制面板的建設(shè)在社區(qū)也沒有提上日程。
融合并支持阿里內(nèi)部

Dubbo 3.0 定下了內(nèi)外融合的策略，也就是說 3.0 的核心最終會(huì)在阿里巴巴的生產(chǎn)系統(tǒng)中部署，相信通過大流量、大規(guī)模的考驗(yàn)，Dubbo 用戶可以獲得一個(gè)性能、穩(wěn)定、服務(wù)治理實(shí)踐各方面俱佳的核心，用戶在生產(chǎn)系統(tǒng)中采用 3.0 也會(huì)更加放心。這一點(diǎn)也是 Dubbo 3.0 最重要的使命。

Filter 鏈的異步化設(shè)計(jì)

Dubbo 最強(qiáng)大的一處設(shè)計(jì)是其在 Filter 鏈上的抽象設(shè)計(jì)，通過其擴(kuò)展機(jī)制的開放性支持，用戶可以對(duì) Dubbo 做功能增強(qiáng)，并允許各個(gè)擴(kuò)展點(diǎn)被定制來是否保留。

Dubbo?的?Filter?定義如下:

@SPI public interface Filter {/*** do invoke filter.* <p>* <code>* // before filter* Result result = invoker.invoke(invocation);* // after filter* return result;* </code>** @param invoker service* @param invocation invocation.* @return invoke result.* @throws RpcException* @see org.apache.dubbo.rpc.Invoker#invoke(Invocation)*/Result invoke(Invoker<?>?invoker,?Invocation invocation) throws RpcException;}

按照“調(diào)用一個(gè)遠(yuǎn)程服務(wù)的方法就像調(diào)用本地的方法一樣”這種說法，這個(gè)直接返回 Result 響應(yīng)的方式是非常好的，用起來是簡(jiǎn)單直接，問題是時(shí)代變換到了需要關(guān)注體驗(yàn)，需要走 Reactive 響應(yīng)式的時(shí)代，也回到基本點(diǎn)：invoke一個(gè) invocation 需要經(jīng)過網(wǎng)絡(luò)在不同的進(jìn)程處理，天然就是異步的過程，也就是發(fā)送請(qǐng)求（invocation）與接收響應(yīng)（Result）本身是兩個(gè)不同的事件，是需要兩個(gè)過程方法來在 Filter 鏈處理。那么如何改造這個(gè)關(guān)鍵的 SPI 呢？有兩種方案：

第一種，把 invoke 的返回值改成 CompletableFuture，好處是一目了然，Result 不在建議同步獲取了；但基礎(chǔ)接口的簽名一改會(huì)導(dǎo)致代碼改造量巨大，同時(shí)也會(huì)讓原有的 SPI 擴(kuò)展不在支持。

第二種，Result 接口直接繼承 CompletationStage，是代表了響應(yīng)的異步計(jì)算。這樣能進(jìn)避免第一種的劣勢(shì)。所以，3.0.0 Preview 版本對(duì)內(nèi)部調(diào)用鏈路實(shí)現(xiàn)做了一次重構(gòu)：基于 CompletableFuture 實(shí)現(xiàn)了框架內(nèi)部的全異步調(diào)用，而在外圍編程上，同時(shí)支持同步、異步調(diào)用模式。

值得注意的是，此次重構(gòu)僅限于框架內(nèi)部實(shí)現(xiàn)，對(duì)使用方?jīng)]有任何影響即接口上保持完全兼容。要了解 Dubbo 異步 API 如何使用，請(qǐng)參考《如何基于 Dubbo 實(shí)現(xiàn)全異步的調(diào)用鏈》（地址：http://dubbo.apache.org/zh-cn/blog/dubbo-new-async.html），這篇文章將著重對(duì)實(shí)現(xiàn)思路和原理做一些簡(jiǎn)單介紹。此次重構(gòu)的要點(diǎn)有：

框架內(nèi)部采用全異步調(diào)用模型，僅在外圍做同步、異步適配；
內(nèi)置Filter鏈支持異步回調(diào)；

基本工作流程

首先我們來看一個(gè)通用的跨網(wǎng)絡(luò)異步調(diào)用的線程模型：

通信框架異步發(fā)送請(qǐng)求消息，請(qǐng)求消息發(fā)送成功后，返回代表業(yè)務(wù)結(jié)果的 CompletableFuture 給業(yè)務(wù)線程。之后對(duì)于 Future 的處理，根據(jù)調(diào)用類型會(huì)有所區(qū)別：

對(duì)于同步請(qǐng)求（如上圖體現(xiàn)的場(chǎng)景），業(yè)務(wù)線程會(huì)調(diào)用 future.get 同步阻塞等待結(jié)果，當(dāng)收到網(wǎng)絡(luò)層返回的業(yè)務(wù)結(jié)果后，future.get 返回并最終將結(jié)果傳遞給調(diào)用發(fā)起方。

對(duì)于異步請(qǐng)求，業(yè)務(wù)線程不會(huì)調(diào)用 future.get，而是將 future 保存在調(diào)用上下文或者直接返回給調(diào)用者，同時(shí)會(huì)為 future 注冊(cè)回調(diào)監(jiān)聽器，以便當(dāng)真正的業(yè)務(wù)結(jié)果從通信層返回時(shí)監(jiān)聽器可以對(duì)結(jié)果做進(jìn)一步的處理。

接下來具體看一下一次異步 Dubbo RPC 請(qǐng)求的調(diào)用流程：

消費(fèi)方面向 Proxy 代理編程，發(fā)出調(diào)用請(qǐng)求，請(qǐng)求經(jīng)過 Filter 鏈向下傳遞。

Invoker.invoke() 將請(qǐng)求異步轉(zhuǎn)發(fā)給網(wǎng)絡(luò)層，并收到代表返回結(jié)果的 Future。

Future 被包裝到 Result，轉(zhuǎn)而由 Result 代表這次遠(yuǎn)程調(diào)用的結(jié)果（由于 Result 的異步屬性，此時(shí)它可能并不包含真正的返回值）。

Result 繼續(xù)沿著調(diào)用鏈返回，在經(jīng)過每個(gè) Filter 時(shí)，Filter 可選擇注冊(cè) Listener 監(jiān)聽器，以便在業(yè)務(wù)結(jié)果返回時(shí)執(zhí)行結(jié)果預(yù)處理。

最終 Proxy 調(diào)用 result.recreate() 將結(jié)果返回給消費(fèi)者：

如果方法是 CompletableFuture 簽名，則返回 Future；
如果方法是普通同步簽名，則返回對(duì)象默認(rèn)值，Future 可通過 RpcContext 拿到；

? 6.?調(diào)用方在拿到代表異步業(yè)務(wù)結(jié)果的 Future 后，可選擇注冊(cè)回調(diào)監(jiān)聽器，以監(jiān)聽真正的業(yè)務(wù)結(jié)果返回。

同步調(diào)用和異步調(diào)用基本上是一致的，并且也是走的回調(diào)模式，只是在鏈路返回之前做了一次阻塞 get 調(diào)用，以確保在收到實(shí)際結(jié)果時(shí)再返回。Filter 在注冊(cè) Listener 時(shí)由于 Future 已處于 complete 狀態(tài)，因此會(huì)同時(shí)觸發(fā)回調(diào) onResponse()/onError()。

關(guān)于流程圖中提到的 Result，Result 在 Dubbo 的一次 RPC 調(diào)用中代表返回結(jié)果，在 3.0 中 Result 自身增加了代表狀態(tài)的接口，類似 Future 現(xiàn)在 Result 可以代表一次未完成的調(diào)用。

要讓 Result 具備代表異步返回結(jié)果的能力，有兩中方式來實(shí)現(xiàn)：

1. Result is a Future，在 Java 8 中更合理的方式是繼承 CompletionStage 接口。

public interface Result extends CompletionStage {}

2. 讓 Result 實(shí)例持有 Future 實(shí)例，與 1 的區(qū)別即是設(shè)計(jì)中選用“繼承”還是“組合”。

public class AsyncRpcResult implements Result {private CompletableFuture<RpcResult> resultFuture;}

同時(shí)，為了讓 Result 更直觀的體現(xiàn)其異步結(jié)果的特性，也為了方便面向 Result 接口編程，我們可以考慮為Result增加一些異步接口：

public interface Result extends Serializable {Result thenApplyWithContext(Function<Result, Result> fn);<U> CompletableFuture<U> thenApply(Function<Result,???extends U> fn);Result get() throws InterruptedException, ExecutionException;}

Filter SPI

Filter 是 Dubbo 預(yù)置的攔截器擴(kuò)展 SPI，用來做請(qǐng)求的預(yù)處理、結(jié)果的后處理，框架本身內(nèi)置了一些攔截器實(shí)現(xiàn)，而從用戶層面，我相信這個(gè) SPI 也應(yīng)該是被擴(kuò)展最多的一個(gè)。在 3.0 版本中，Filter 回歸單一職責(zé)的設(shè)計(jì)模式，將回調(diào)接口單獨(dú)提取到 Listener 中。

@SPI public interface Filter {Result invoke(Invoker<?>?invoker,?Invocation invocation) throws RpcException;interface Listener {void onResponse(Result result, Invoker<?>?invoker,?Invocation invocation);void onError(Throwable t, Invoker<?>?invoker,?Invocation invocation);} }

以上是 Filter 的 SPI 定義，Filter 的核心定義中只有一個(gè) invoke() 方法用來傳遞調(diào)用請(qǐng)求。

同時(shí)，增加了一個(gè)新的回調(diào)接口 Listener，每個(gè) Filter 實(shí)現(xiàn)可以定義自己的 Listenr 回調(diào)器，從而實(shí)現(xiàn)對(duì)返回結(jié)果的異步監(jiān)聽，參考以下是為 MonitorFilter 增加的 Listener 回調(diào)實(shí)現(xiàn)：

class MonitorListener implements Listener {@Overridepublic void onResponse(Result result, Invoker<?>?invoker,?Invocation invocation) {if (invoker.getUrl().hasParameter(Constants.MONITOR_KEY)) {collect(invoker, invocation, result, RpcContext.getContext().getRemoteHost(),Long.valueOf(invocation.getAttachment(MONITOR_FILTER_START_TIME)), false);getConcurrent(invoker, invocation).decrementAndGet(); // count down}}@Overridepublic void onError(Throwable t, Invoker<?>?invoker,?Invocation invocation) {if (invoker.getUrl().hasParameter(Constants.MONITOR_KEY)) {collect(invoker, invocation, null, RpcContext.getContext().getRemoteHost(), Long.valueOf(invocation.getAttachment(MONITOR_FILTER_START_TIME)), true);getConcurrent(invoker, invocation).decrementAndGet(); // count down}} }

泛化調(diào)用異步接口支持

為了更直觀的做異步調(diào)用，泛化接口新增了?

CompletableFuture<Object>$invokeAsync(Stringmethod,String[]parameterTypes,Object[]args)接口：

public interface GenericService {/*** Generic invocation** @param method Method name, e.g. findPerson. If there are overridden methods, parameter info is* required, e.g. findPerson(java.lang.String)* @param parameterTypes Parameter types* @param args Arguments* @return invocation return value* @throws GenericException potential exception thrown from the invocation*/Object $invoke(String method, String[] parameterTypes, Object[] args) throws GenericException;default CompletableFuture<Object> $invokeAsync(String method, String[] parameterTypes, Object[] args) throws GenericException {Object object = $invoke(method, parameterTypes, args);if (object instanceof CompletableFuture) {return (CompletableFuture<Object>) object;}return CompletableFuture.completedFuture(object);} }

這樣，當(dāng)我們想做異步調(diào)用時(shí)，就可以直接這樣使用：

CompletableFuture<Object> genericService.$invokeAsync(method, parameterTypes, args);

更具體用例請(qǐng)參見《泛化調(diào)用示例》

https://github.com/apache/incubator-dubbo-samples/tree/3.x/dubbo-samples-generic/dubbo-samples-generic-call

異步與性能

組要注意的是，框架內(nèi)部的異步實(shí)現(xiàn)本身并不能提高單次調(diào)用的性能，相反，由于線程切換和回調(diào)邏輯的存在，異步反而可能會(huì)導(dǎo)致單次調(diào)用性能的下降，但是異步帶來的優(yōu)勢(shì)是能減少對(duì)資源的占用，提升整個(gè)系統(tǒng)的并發(fā)程度和吞吐量，這點(diǎn)對(duì)于 RPC 這種需要處理網(wǎng)絡(luò)延遲的場(chǎng)景非常適用。更多關(guān)于異步化設(shè)計(jì)的好處，請(qǐng)參考其他異步化原理介紹相關(guān)文章。

響應(yīng)式編程支持

響應(yīng)式編程讓開發(fā)者更方便地編寫高性能的異步代碼，很可惜，在之前很長(zhǎng)一段時(shí)間里，dubbo 并不支持響應(yīng)式編程，簡(jiǎn)單來說，dubbo 不支持在 rpc 調(diào)用時(shí)使用 Mono/Flux 這種流對(duì)象（reative-stream 里流的概念)，給用戶使用帶來了不便。（關(guān)于響應(yīng)式編程更詳細(xì)的信息請(qǐng)參見這里：http://reactivex.io/)。

RSocket 是一個(gè)開源的支持 reactive-stream 語義的網(wǎng)絡(luò)通信協(xié)議，他將 reative 語義的復(fù)雜邏輯封裝起來了，使得上層可以方便實(shí)現(xiàn)網(wǎng)絡(luò)程序。（RSocket詳細(xì)資料請(qǐng)參見這里：http://rsocket.io/)。

dubbo 在 3.0.0-SNAPSHOT ?版本里基于 RSocket 對(duì)響應(yīng)式編程進(jìn)行了簡(jiǎn)單的支持，用戶可以在請(qǐng)求參數(shù)和返回值里使用 Mono 和 Flux 類型的對(duì)象。下面我們給出使用范例，（范例源碼可以在這里獲取：https://github.com/apache/incubator-dubbo-samples/tree/3.x/dubbo-samples-rsocket)。

首先定義接口如下：

public interface DemoService {Mono<String> requestMonoWithMonoArg(Mono<String> m1, Mono<String> m2);Flux<String> requestFluxWithFluxArg(Flux<String> f1, Flux<String> f2); }

然后實(shí)現(xiàn)該 demo 接口：

public class DemoServiceImpl implements DemoService {@Overridepublic Mono<String> requestMonoWithMonoArg(Mono<String> m1, Mono<String> m2) {return m1.zipWith(m2, new BiFunction<String, String, String>() {@Overridepublic String apply(String s, String s2) {return s+" "+s2;}});}@Overridepublic Flux<String> requestFluxWithFluxArg(Flux<String> f1, Flux<String> f2) {return f1.zipWith(f2, new BiFunction<String, String, String>() {@Overridepublic String apply(String s, String s2) {return s+" "+s2;}});} }

然后配置并啟動(dòng)服務(wù)端，注意協(xié)議名字填寫 rsocket：

<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.xsdhttp://dubbo.apache.org/schema/dubbo?http://dubbo.apache.org/schema/dubbo/dubbo.xsd">  <dubbo:application name="demo-provider"/>  <dubbo:registry address="zookeeper://127.0.0.1:2181"/>  <dubbo:protocol name="rsocket" port="20890"/>  <bean id="demoService" class="org.apache.dubbo.samples.basic.impl.DemoServiceImpl"/>  <dubbo:service interface="org.apache.dubbo.samples.basic.api.DemoService" ref="demoService"/> </beans> public class RsocketProvider {public static void main(String[] args) throws Exception {new EmbeddedZooKeeper(2181, false).start();ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[]{"spring/rsocket-provider.xml"});context.start();System.in.read(); // press any key to exit} }

然后配置并啟動(dòng)消費(fèi)者消費(fèi)者如下, 注意協(xié)議名填寫 rsocket：

<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.xsdhttp://dubbo.apache.org/schema/dubbo?http://dubbo.apache.org/schema/dubbo/dubbo.xsd">  <dubbo:application name="demo-consumer"/>  <dubbo:registry address="zookeeper://127.0.0.1:2181"/>  <dubbo:reference id="demoService" check="true" interface="org.apache.dubbo.samples.basic.api.DemoService"/> </beans> public class RsocketConsumer {public static void main(String[] args) {ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[]{"spring/rsocket-consumer.xml"});context.start();DemoService demoService = (DemoService) context.getBean("demoService"); // get remote service proxywhile (true) {try {Mono<String> monoResult = demoService.requestMonoWithMonoArg(Mono.just("A"), Mono.just("B"));monoResult.doOnNext(new Consumer<String>() {@Overridepublic void accept(String s) {System.out.println(s);}}).block();Flux<String> fluxResult = demoService.requestFluxWithFluxArg(Flux.just("A","B","C"), Flux.just("1","2","3"));fluxResult.doOnNext(new Consumer<String>() {@Overridepublic void accept(String s) {System.out.println(s);}}).blockLast();} catch (Throwable throwable) {throwable.printStackTrace();}}} }

可以看到配置上除了協(xié)議名使用 rsocket 以外其他并沒有特殊之處。

實(shí)現(xiàn)原理

以前用戶并不能在參數(shù)或者返回值里使用 Mono/Flux 這種流對(duì)象（reative-stream 里的流的概念）。因?yàn)榱鲗?duì)象自帶異步屬性，當(dāng)業(yè)務(wù)把流對(duì)象作為參數(shù)或者返回值傳遞給框架之后，框架并不能將流對(duì)象正確的進(jìn)行序列化。

dubbo 基于 RSocket 實(shí)現(xiàn)了 reative 支持。RSocket 將 reative 語義的復(fù)雜邏輯封裝起來了，給上層提供了簡(jiǎn)潔的抽象如下：

/** * Fire and Forget interaction model of {@code RSocket}. * * @param payload Request payload. * @return {@code Publisher} that completes when the passed {@code payload} is successfully * handled, otherwise errors. */ Mono<Void> fireAndForget(Payload payload); /** * Request-Response interaction model of {@code RSocket}. * * @param payload Request payload. * @return {@code Publisher} containing at most a single {@code Payload} representing the * response. */ Mono<Payload> requestResponse(Payload payload); /** * Request-Stream interaction model of {@code RSocket}. * * @param payload Request payload. * @return {@code Publisher} containing the stream of {@code Payload}s representing the response. */ Flux<Payload> requestStream(Payload payload); /** * Request-Channel interaction model of {@code RSocket}. * * @param payloads Stream of request payloads. * @return Stream of response payloads. */ Flux<Payload> requestChannel(Publisher<Payload> payloads);

我們只需要在此基礎(chǔ)上添加我們的 rpc 邏輯即可。

從客戶端視角看，框架建立連接之后，只需要將請(qǐng)求信息編碼到 Payload 里，然后通過 requestStream 方法即可向服務(wù)端發(fā)起請(qǐng)求。
從服務(wù)端視角看，rsocket 收到請(qǐng)求之后，會(huì)調(diào)用我們實(shí)現(xiàn)的 requestStream 方法，我們從 Payload 里解碼得到請(qǐng)求信息之后，調(diào)用業(yè)務(wù)方法，然后拿到 Flux 類型的返回值即可。
需要注意的是業(yè)務(wù)返回值一般是 Flux，而 RSocket 要求的是 Flux，所以我們需要通過 map operator 攔截業(yè)務(wù)數(shù)據(jù)，將 BizDO 編碼為 Payload 才可以遞交給我 RSocket。而 RSocket 會(huì)負(fù)責(zé)數(shù)據(jù)的傳輸和 reative 語義的實(shí)現(xiàn)。

經(jīng)過上面的分析，我們知道了 Dubbo 如何基于 RSocket 實(shí)現(xiàn)了響應(yīng)式編程的支持。有了響應(yīng)式編程支持，業(yè)務(wù)可以更加方便的實(shí)現(xiàn)異步邏輯。

小結(jié)

當(dāng)前 Dubbo 3.0 將提供具備當(dāng)代特性（如響應(yīng)性編程）的相關(guān)支持，同時(shí)汲取阿里內(nèi)部 HSF 的設(shè)計(jì)長(zhǎng)處來實(shí)現(xiàn)兩者的融合，當(dāng)前預(yù)覽版的很多地方還在探討中，希望大家能夠積極反饋，我們都會(huì)虛心學(xué)習(xí)并參考。

Dubbo 3.0 sample @GitHub：

https://github.com/apache/incubator-dubbo-samples/tree/3.x

覃柳杰（花名：未宇） Github ID: qinliujie，阿里巴巴中間件開發(fā)，Dubbo 開源項(xiàng)目 PMC，參與 HSF2.2和 Dubbo3.0 的設(shè)計(jì)和開發(fā)。

呂仁琦（花名：空冥） Github ID: jefflv，阿里巴巴中間件開發(fā)，Dubbo 開源項(xiàng)目 commiter，參與了內(nèi)部 HSF2.0 的設(shè)計(jì)和開發(fā)。

劉軍（花名：陸龜）阿里巴巴中間件高級(jí)開發(fā)工程師，Apache Dubbo (Incubating)PPMC，深度參與 Dubbo 項(xiàng)目開發(fā)，主要貢獻(xiàn)者之一。

謝育能（花名：思邪）阿里巴巴中間件開發(fā)，Dubbo 3.0 開源項(xiàng)目的響應(yīng)式模塊的負(fù)責(zé)人，參與了內(nèi)部 HSF2.2 的設(shè)計(jì)和開發(fā)。

總結(jié)

以上是生活随笔為你收集整理的Dubbo 3.0 预览版解读，6到飞起~的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。

上一篇：欠阿里云一分钱，会是什么样的后果。。。
下一篇： Java 程序员必备的 Intellij