作者 | 雷卷
來(lái)源|阿里巴巴云原生公眾號(hào)

RSocket 分布式通訊協(xié)議是 Spring Reactive 的核心內(nèi)容，從 Spring Framework 5.2 開(kāi)始，RSocket 已經(jīng)是 Spring 的內(nèi)置功能，Spring Boot 2.3 也添加了 spring-boot-starter-rsocket，簡(jiǎn)化了 RSocket 的服務(wù)編寫和服務(wù)調(diào)用。RSocket 通訊的核心架構(gòu)中包含兩種模式，分別是 Broker 代理模式和服務(wù)直連通訊模式。

Broker 的通訊模式更靈活，如 Alibaba RSocket Broker，采用的是事件驅(qū)動(dòng)模型架構(gòu)。而目前更多的架構(gòu)則是面向服務(wù)化設(shè)計(jì)，也就是我們常說(shuō)的服務(wù)注冊(cè)發(fā)現(xiàn)和服務(wù)直連通訊的模式，其中最知名的就是 Spring Cloud 技術(shù)棧，涉及到配置推送、服務(wù)注冊(cè)發(fā)現(xiàn)、服務(wù)網(wǎng)關(guān)、斷流保護(hù)等等。在面向服務(wù)化的分布式網(wǎng)絡(luò)通訊中，如 REST API、gRPC 和 Alibaba Dubbo 等，都與 Spring Cloud 有很好地集成，用戶基本不用關(guān)心服務(wù)注冊(cè)發(fā)現(xiàn)和客戶端負(fù)載均衡這些底層細(xì)節(jié)，就可以完成非常穩(wěn)定的分布式網(wǎng)絡(luò)通訊架構(gòu)。

RSocket 作為通訊協(xié)議的后起之秀，核心是二進(jìn)制異步化消息通訊，是否也能和 Spring Cloud 技術(shù)棧結(jié)合，實(shí)現(xiàn)服務(wù)注冊(cè)發(fā)現(xiàn)、客戶端負(fù)載均衡，從而更高效地實(shí)現(xiàn)面向服務(wù)的架構(gòu)？這篇文章我們就討論一下 Spring Cloud 和 RSocket 結(jié)合實(shí)現(xiàn)服務(wù)注冊(cè)發(fā)現(xiàn)和負(fù)載均衡。

服務(wù)注冊(cè)發(fā)現(xiàn)

服務(wù)注冊(cè)發(fā)現(xiàn)的原理非常簡(jiǎn)單，主要涉及三種角色：服務(wù)提供方、服務(wù)消費(fèi)者和服務(wù)注冊(cè)中心。典型的架構(gòu)如下：

服務(wù)提供方，如 RSocket Server，在應(yīng)用啟動(dòng)后，會(huì)向服務(wù)注冊(cè)中心注冊(cè)應(yīng)用相關(guān)的信息，如應(yīng)用名稱，ip 地址，Web Server 監(jiān)聽(tīng)端口號(hào)等，當(dāng)然還會(huì)包括一些元信息，如服務(wù)的分組(group)，服務(wù)的版本號(hào)(version)，RSocket 的監(jiān)聽(tīng)端口號(hào)，如果是 WebSocket 通訊，還需要提供 ws 映射路徑等，不少開(kāi)發(fā)者會(huì)將服務(wù)提供方的服務(wù)接口列表作為 tags 提交給服務(wù)注冊(cè)中心，方便后續(xù)的服務(wù)查詢和治理。

在本文中，我們采用 Consul 作為服務(wù)注冊(cè)中心，主要是 Consul 比較簡(jiǎn)單，下載后執(zhí)行 consul agent -dev 就可以啟動(dòng)對(duì)應(yīng)的服務(wù)，當(dāng)然你可以使用 Docker Compose，配置也非常簡(jiǎn)單，然后 docker-compose up -d 就可以啟動(dòng) Consul 服務(wù)。

當(dāng)我們向服務(wù)中心注冊(cè)和查詢服務(wù)時(shí)，都需要有一個(gè)應(yīng)用名稱，對(duì)應(yīng)到 Spring Cloud 中，也就是 Spring Boot 對(duì)應(yīng)的 spring.application.name 的值，這里我們稱之為應(yīng)用名稱，也就是后續(xù)的服務(wù)查找都是基于該應(yīng)用名稱進(jìn)行的。如果你調(diào)用 ReactiveDiscoveryClient.getInstances(String serviceId); 查找服務(wù)實(shí)例列表時(shí)，這個(gè) serviceId 參數(shù)其實(shí)就是 Spring Boot 的應(yīng)用名稱。考慮到服務(wù)注冊(cè)和后續(xù)的 RSocket 服務(wù)路由的配合以及方便大家理解，這里我們打算設(shè)計(jì)一個(gè)簡(jiǎn)單的命名規(guī)范。

假設(shè)你有一個(gè)服務(wù)應(yīng)用，功能名稱為 calculator，同時(shí)提供兩個(gè)服務(wù): 數(shù)學(xué)計(jì)算器服務(wù)(MathCalculatorService)和匯率計(jì)算器服務(wù)(ExchangeCalculatorService), ?那么我們?cè)撊绾蝸?lái)命名該應(yīng)用及其對(duì)應(yīng)的服務(wù)接口名？

這里我們采用類似 Java package 命名規(guī)范，采用域名倒排的方式，如 calculator 應(yīng)用對(duì)應(yīng)的則為 com-example-calculator 樣式，為何是中劃線，而不是點(diǎn)？. 在 DNS 解析中作為主機(jī)名是非法的，只能作為子域名存在，不能作為主機(jī)名，而目前的服務(wù)注冊(cè)中心設(shè)計(jì)都遵循 DNS 規(guī)約，所以我們采用中劃線的方式來(lái)命名應(yīng)用。這樣采用域名倒排和應(yīng)用名結(jié)合的方式，可以確保應(yīng)用之間不會(huì)重名，另外也方便和 Java Package 名稱進(jìn)行轉(zhuǎn)換，也就是 -? 和 . 之間的相互轉(zhuǎn)換。

那么應(yīng)用包含的服務(wù)接口應(yīng)該如何命名？服務(wù)接口全名是由應(yīng)用名稱和 interface 名稱組合而成，規(guī)則如下：

String serviceFullName = appName.replace("-", ".") + "." + serviceInterfaceName;

例如以下的服務(wù)命名都是合乎規(guī)范的：

• com.example.calculator.MathCalculatorService

• com.example.calculator.ExchangeCalculatorService

而 com.example.calculator.math.MathCalculatorService 則是錯(cuò)誤的, ?因?yàn)樵趹?yīng)用名稱和接口名稱之間多了 math。為何要采用這種命名規(guī)范？首先讓我們看一下服務(wù)消費(fèi)方是如何調(diào)用遠(yuǎn)程服務(wù)的。假設(shè)服務(wù)消費(fèi)方拿到一個(gè)服務(wù)接口，如 com.example.calculator.MathCalculatorService，那么他該如何發(fā)起服務(wù)調(diào)用呢？

• 首先根據(jù) Service 全面提取處對(duì)應(yīng)的應(yīng)用名稱(appName)，如 com.example.calculator.MathCalculatorService 服務(wù)對(duì)應(yīng)的 appName 則為 com-example-calculator。如果應(yīng)用和服務(wù)接口之間不存在任何關(guān)系，那么想要獲取服務(wù)接口對(duì)應(yīng)的服務(wù)提供方信息，你可能還需要應(yīng)用名稱，這會(huì)相對(duì)來(lái)說(shuō)比較麻煩。如果接口名稱中包含對(duì)應(yīng)的應(yīng)用信息，則會(huì)簡(jiǎn)單很多，你可以理解為應(yīng)用是服務(wù)全面中的一部分。

• 調(diào)用 ReactiveDiscoveryClient.getInstances(appName) 獲取應(yīng)用名對(duì)應(yīng)的服務(wù)實(shí)例列表(ServiceInstance)，ServiceInstance 對(duì)象會(huì)包含諸如 IP 地址，Web 端口號(hào)、RSocket 監(jiān)聽(tīng)端口號(hào)等其他元信息。

• 根據(jù) RSocketRequester.Builder.transports(servers) 構(gòu)建具有負(fù)載均衡能力的 RSocketRequester 對(duì)象。

• 使用服務(wù)全稱和具體功能名稱作為路由進(jìn)行 RSocketRequester 的 API 調(diào)用，樣例代碼如下：

rsocketRequester .route("com.example.calculator.MathCalculatorService.square") ?.data(number) ?.retrieveMono(Integer.class)

通過(guò)上述的命名規(guī)范，我們可以從服務(wù)接口全稱中提取出應(yīng)用名，然后和服務(wù)注冊(cè)中心交互查找對(duì)應(yīng)的實(shí)例列表，然后建立和服務(wù)提供者的連接，最后基于服務(wù)名稱進(jìn)行服務(wù)調(diào)用。該命名規(guī)范，基本做到到了最小化的依賴，開(kāi)發(fā)者完全是基于服務(wù)接口調(diào)用，非常簡(jiǎn)單。

RSocket 服務(wù)編寫

有了服務(wù)的命名規(guī)范和服務(wù)注冊(cè)，編寫 RSocket 服務(wù)，這個(gè)還是非常簡(jiǎn)單，和編寫一個(gè) Spring Bean 沒(méi)有任何區(qū)別。引入 spring-boot-starter-rsocket 依賴，創(chuàng)建一個(gè) Controller 類，添加對(duì)應(yīng)的 MessagMapping annotation 作為基礎(chǔ)路由，然后實(shí)現(xiàn)功能接口添加功能名稱，樣例代碼如下：

@Controller @MessageMapping("com.example.calculator.MathCalculatorService") public class MathCalculatorController implements MathCalculatorService { ? ? @MessageMapping("square") ? ? public Mono<Integer> square(Integer input) { ? ? ? ? System.out.println("received: " + input); ? ? ? ? return Mono.just(input * input); ? ? } }

上述代碼看起來(lái)好像有點(diǎn)奇怪，既然是服務(wù)實(shí)現(xiàn)，添加 @Controller 和 @MessageMapping，看起來(lái)好像有點(diǎn)不倫不類的。當(dāng)然這些 annotation 都是一些技術(shù)細(xì)節(jié)體現(xiàn)，你也能看出，RSocket 的服務(wù)實(shí)現(xiàn)是基于 Spring Message 的，是面向消息化的。這里我們其實(shí)只需要添加一個(gè)自定義的 @SpringRSocketService annotation 就可以解決這個(gè)問(wèn)題，代碼如下：

@Target(ElementType.TYPE) @Retention(RetentionPolicy.RUNTIME) @Documented @Controller @MessageMapping() public @interface SpringRSocketService { ? ? @AliasFor(annotation = MessageMapping.class) ? ? String[] value() default {}; }

回到服務(wù)對(duì)應(yīng)的實(shí)現(xiàn)代碼，我們改為使用 @SpringRSocketService annotation，這樣我們的代碼就和標(biāo)準(zhǔn)的 RPC 服務(wù)接口完全一模一樣啦，也便于理解。此外 @SpringRSocketService 和 @RSocketHandler 這兩個(gè) Annotation，也方便我們后續(xù)做一些 Bean 掃描、IDE 插件輔助等。

@SpringRSocketService("com.example.calculator.MathCalculatorService") public class MathCalculatorImpl implements MathCalculatorService { ? ? @RSocketHandler("square") ? ? public Mono<Integer> square(Integer input) { ? ? ? ? System.out.println("received: " + input); ? ? ? ? return Mono.just(input * input); ? ? } }

最后我們添加一下 spring-cloud-starter-consul-discovery 依賴，設(shè)置一下 bootstrap.properties，然后在 application.properties 設(shè)置一下 RSocket 監(jiān)聽(tīng)的端口和元信息，我們還將該應(yīng)用提供的服務(wù)接口列表作為 tags 傳給服務(wù)注冊(cè)中心，當(dāng)然這個(gè)也是方便我們后續(xù)的服務(wù)管理。樣例如下：

spring.application.name=com-example-calculator spring.cloud.consul.discovery.instance-id=com-example-calculator-${random.uuid} spring.cloud.consul.discovery.prefer-ip-address=true server.port=0 spring.rsocket.server.port=6565 spring.cloud.consul.discovery.metadata.rsocketPort=${spring.rsocket.server.port} spring.cloud.consul.discovery.tags=com.example.calculator.ExchangeCalculatorService,com.example.calculator.MathCalculatorService

RSocket 服務(wù)應(yīng)用啟動(dòng)后，我們?cè)?Consul 控制臺(tái)就可以看到服務(wù)注冊(cè)上來(lái)的信息，截屏如下：

RSocket 客戶端接入

客戶端接入稍微有一點(diǎn)復(fù)雜，主要是要基于服務(wù)接口全面要做一系列相關(guān)的操作，但是前面我們已經(jīng)有了命名規(guī)范，所以問(wèn)題也不大。客戶端應(yīng)用同樣會(huì)接入服務(wù)注冊(cè)中心，這樣我們就可以獲得 ReactiveDiscoveryClient bean，接下來(lái)就是根據(jù)服務(wù)接口全名，如 com.example.calculator.ExchangeCalculatorService 構(gòu)建出具有負(fù)載均衡的 RSocketRequester。

原理也非常簡(jiǎn)單，前面說(shuō)過(guò)，根據(jù)服務(wù)接口全稱，獲得其對(duì)應(yīng)的應(yīng)用名稱，然后調(diào)用 ReactiveDiscoveryClient.getInstances(appName) 獲得服務(wù)應(yīng)用對(duì)應(yīng)的實(shí)例列表，接下來(lái)將服務(wù)實(shí)例(ServiceInstance）列表轉(zhuǎn)換為 RSockt 的 LoadbalanceTarget 列表，其實(shí)就是 POJO 轉(zhuǎn)換，最后將轉(zhuǎn) LoadbalanceTarget 列表進(jìn)行 Flux 封裝(如使用 Sink 接口)，傳遞給 RSocketRequester.Builder 就完成具有負(fù)載均衡能力的 RSocketRequester 構(gòu)建，詳細(xì)的代碼細(xì)節(jié)大家可以參考項(xiàng)目的代碼庫(kù)。

這里要注意的是接下來(lái)如何感知服務(wù)端實(shí)例列表的變化，如應(yīng)用上下線，服務(wù)暫停等。這里我采用一個(gè)定時(shí)任務(wù)方案，定時(shí)查詢服務(wù)對(duì)應(yīng)的地址列表。當(dāng)然還有其他的機(jī)制，如果是標(biāo)準(zhǔn)的 Spring Cloud 服務(wù)發(fā)現(xiàn)接口，目前是需要客戶端輪詢的，當(dāng)然也可以結(jié)合 Spring Cloud Bus 或者消息中間件，實(shí)現(xiàn)服務(wù)端列表變化的監(jiān)聽(tīng)。如果客戶端感知到服務(wù)列表的變化，只需要調(diào)用 Reactor 的 Sink 接口發(fā)送新的列表即可，RSocket Load Balance 在感知到變化后，會(huì)自動(dòng)做出響應(yīng)，如關(guān)閉即將失效的連接、創(chuàng)建新的連接等工作。

在實(shí)際的應(yīng)用之間的相互通訊，會(huì)存在一些服務(wù)提供方不可用的情況，如服務(wù)方突然宕機(jī)或者其網(wǎng)絡(luò)不可用，這就導(dǎo)致了服務(wù)應(yīng)用列表中部分服務(wù)不可用，那么 RSocket 這個(gè)時(shí)候會(huì)如何處理？不用擔(dān)心，RSocket Load Balance 有重試機(jī)制，當(dāng)一個(gè)服務(wù)調(diào)用出現(xiàn)連接等異常，會(huì)重新從列表中獲取一個(gè)連接進(jìn)行通訊，而那個(gè)錯(cuò)誤的連接也會(huì)標(biāo)識(shí)為可用性為 0，不會(huì)再被后續(xù)請(qǐng)求所使用。服務(wù)列表推送和通訊期間的容錯(cuò)重試機(jī)制，這兩者保證了分布式通訊的高可用性。

最后讓我們啟動(dòng) client-app，然后從客戶端發(fā)起一個(gè)遠(yuǎn)程的 RSocket 調(diào)用，截屏如下：

上圖中 com-example-calculator 服務(wù)應(yīng)用包括三個(gè)實(shí)例，服務(wù)的調(diào)用會(huì)在這三個(gè)服務(wù)實(shí)例交替進(jìn)行(RoundRobin 策略)。

開(kāi)發(fā)體驗(yàn)的一些考量

雖然服務(wù)注冊(cè)和發(fā)現(xiàn)、客戶端的負(fù)載均衡這些都完成啦，調(diào)用和容錯(cuò)這些都沒(méi)有問(wèn)題，但是還有一些使用體驗(yàn)上的問(wèn)題，這里我們也闡述一下，讓開(kāi)發(fā)體驗(yàn)做的更好。

1. 基于服務(wù)接口通訊

大多數(shù) RPC 通訊都是基于接口的，如 Apache Dubbo、gRPC 等。那么 RSocket 能否做到？答案是其實(shí)完全可以。在服務(wù)端，我們已經(jīng)是基于服務(wù)接口來(lái)實(shí)現(xiàn) RSocket 服務(wù)啦，接下來(lái)我們只需要在客戶端實(shí)現(xiàn)基于該接口的調(diào)用就可以。對(duì)于 Java 開(kāi)發(fā)者來(lái)說(shuō)，這不是大問(wèn)題，我們只需要基于 Java Proxy 機(jī)制構(gòu)建就可以，而 Proxy 對(duì)應(yīng)的 InvocationHandler 會(huì)使用 RSocketRequester 來(lái)實(shí)現(xiàn) invoke() 的函數(shù)調(diào)用。詳細(xì)的細(xì)節(jié)請(qǐng)參考應(yīng)用代碼中的的 RSocketRemoteServiceBuilder.java 文件，而且在 client-app module 中也已經(jīng)包含了解基于接口調(diào)用的 bean 實(shí)現(xiàn)。

2. 服務(wù)接口函數(shù)的單參數(shù)問(wèn)題

使用 RSocketRequester 調(diào)用遠(yuǎn)程接口時(shí)，對(duì)應(yīng)的處理函數(shù)只能接受單個(gè)參數(shù)，這個(gè)和 gRPC 的設(shè)計(jì)是類似的，當(dāng)然也考慮了不同對(duì)象序列化框架的支持問(wèn)題。但是考慮到實(shí)際的使用體驗(yàn)，可能會(huì)涉及到多參函數(shù)的情況，讓調(diào)用方開(kāi)發(fā)體驗(yàn)更好，那么這個(gè)時(shí)候該如何處理？其實(shí)從 Java 1.8 后，interface 是允許增加 default 函數(shù)的，我們可以添加一些體驗(yàn)更友好的 default 函數(shù)，而且還不影響服務(wù)通訊接口，樣例如下：

public interface ExchangeCalculatorService { ? ? double exchange(ExchangeRequest request); ? ? default double rmbToDollar(double amount) { ? ? ? ? return exchange(new ExchangeRequest(amount, "CNY", "USD")); ? ? } }

通過(guò) interface 的 default method，我們可以為調(diào)用方提供給便捷函數(shù)，如在網(wǎng)絡(luò)傳輸?shù)氖亲止?jié)數(shù)組 (byte[])，但是在 default 函數(shù)中，我們可以添加 File 對(duì)象支持，方便調(diào)用方使用。Interface 中的函數(shù) API 負(fù)責(zé)服務(wù)通訊規(guī)約，default 函數(shù)來(lái)提升使用方的體驗(yàn)，這兩者的配合，可以非常容易解決函數(shù)多參問(wèn)題，當(dāng)然 default 函數(shù)在一定程度上還可以作為數(shù)據(jù)驗(yàn)證的前哨來(lái)使用。

3. RSocket Broker 支持

前面我們說(shuō)到，RSocket 還有一種 Broker 架構(gòu)，也就是服務(wù)提供方是隱藏在 Broker 之后的，請(qǐng)求主要是由 Broker 承接，然后再轉(zhuǎn)發(fā)給服務(wù)提供方處理，架構(gòu)樣例如下：

那么基于服務(wù)發(fā)現(xiàn)的機(jī)制負(fù)載均衡，能否和 RSocket Broker 模式混合使用呢？如一些長(zhǎng)尾或者復(fù)雜網(wǎng)絡(luò)下的應(yīng)用，可以注冊(cè)到 RSocket Broker，然后由 Broker 處理請(qǐng)求調(diào)用和轉(zhuǎn)發(fā)。這個(gè)其實(shí)也不不復(fù)雜，前面我們說(shuō)到應(yīng)用和服務(wù)接口命名規(guī)范，這里我們只需要添加一個(gè)應(yīng)用名前綴就可以解決。假設(shè)我們有一個(gè) RSocker Broker 集群，暫且我們稱之為 broker0 集群，當(dāng)然該 broker 集群的實(shí)例也都注冊(cè)到服務(wù)注冊(cè)中心（如 Consul）啦。那么在調(diào)用 RSocket Broker 上的服務(wù)時(shí)，服務(wù)名稱就被調(diào)整為 broker0:com.example.calculator.MathCalculatorService，也就是服務(wù)名前添加了 appName: 這樣的前綴，這個(gè)其實(shí)是 URI 的另一種規(guī)范形式，我們就可以提取冒號(hào)之前的應(yīng)用名，然后去服務(wù)注冊(cè)中心查詢獲得應(yīng)用對(duì)應(yīng)的實(shí)例列表。

回到 Broker 互通的場(chǎng)景，我們會(huì)向服務(wù)注冊(cè)中心查詢 broker0 對(duì)應(yīng)的服務(wù)列表，然后和 broker0 集群的實(shí)例列表創(chuàng)建連接，這樣后續(xù)基于該接口的服務(wù)調(diào)用就會(huì)發(fā)送給 Broker 進(jìn)行處理，也就是完成了服務(wù)注冊(cè)發(fā)現(xiàn)和 Broker 模式的混合使用的模式。

借助于這種定向指定服務(wù)接口和應(yīng)用間的關(guān)聯(lián)，也方便我們做一些 beta 測(cè)試，如你想將 com.example.calculator.MathCalculatorService 的調(diào)用導(dǎo)流到 beta 應(yīng)用，你就可以使用 com-example-calculator-beta1:com.example.calculator.MathCalculatorService 這種方式調(diào)用服務(wù)，這樣服務(wù)調(diào)用對(duì)應(yīng)的流量就會(huì)轉(zhuǎn)發(fā)給 com-example-calculator-beta1 對(duì)應(yīng)的實(shí)例，起到 beta 測(cè)試的效果。

回到最前面說(shuō)到的規(guī)范，如果應(yīng)用名和服務(wù)接口的綁定關(guān)系你實(shí)在做不到，那么你可以使用這種方式實(shí)現(xiàn)服務(wù)調(diào)用，如 calculator-server:com.example.calculator.math.MathCalculatorService，只是你需要更完整的文檔說(shuō)明，當(dāng)然這種方式也可以解決之前系統(tǒng)接入到目前的架構(gòu)上，應(yīng)用的遷移成本也比較小。如果你之前的面向服務(wù)化架構(gòu)設(shè)計(jì)也是基于 interface 接口通訊的，那么通過(guò)該方式遷移到 RSocket 上完全沒(méi)有問(wèn)題，對(duì)客戶端代碼調(diào)整也最小。

總結(jié)

通過(guò)整合服務(wù)注冊(cè)發(fā)現(xiàn)，結(jié)合一個(gè)實(shí)際的命名規(guī)范，就完成了服務(wù)注冊(cè)發(fā)現(xiàn)和 RSocket 路由之間的優(yōu)雅配合，當(dāng)然負(fù)載均衡也是包含其中啦。對(duì)比其他的 RPC 方案，你不需要引入 RPC 自己的服務(wù)注冊(cè)中心，復(fù)用 Spring Cloud 的服務(wù)注冊(cè)中心就可以，如 Alibaba Nacos， Consul， Eureka 和 ZooKeeper 等，沒(méi)有多余的開(kāi)銷和維護(hù)成本。如果你想更多了解 RSocket RPC 相關(guān)的細(xì)節(jié)，可以參考 Spring 官方博客?《Easy RPC with RSocket》。

歡迎加入 alibaba-rsocket-broker 釘釘群：

更多詳細(xì)的代碼細(xì)節(jié)，可以點(diǎn)擊鏈接查看文章對(duì)應(yīng)的代碼庫(kù)！

總結(jié)

以上是生活随笔為你收集整理的Spring RSocket：基于服务注册发现的 RSocket 负载均衡的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

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