作者 | 雷卷
來源|阿里巴巴云原生公眾號

RSocket 分布式通訊協議是 Spring Reactive 的核心內容，從 Spring Framework 5.2 開始，RSocket 已經是 Spring 的內置功能，Spring Boot 2.3 也添加了 spring-boot-starter-rsocket，簡化了 RSocket 的服務編寫和服務調用。RSocket 通訊的核心架構中包含兩種模式，分別是 Broker 代理模式和服務直連通訊模式。

Broker 的通訊模式更靈活，如 Alibaba RSocket Broker，采用的是事件驅動模型架構。而目前更多的架構則是面向服務化設計，也就是我們常說的服務注冊發現和服務直連通訊的模式，其中最知名的就是 Spring Cloud 技術棧，涉及到配置推送、服務注冊發現、服務網關、斷流保護等等。在面向服務化的分布式網絡通訊中，如 REST API、gRPC 和 Alibaba Dubbo 等，都與 Spring Cloud 有很好地集成，用戶基本不用關心服務注冊發現和客戶端負載均衡這些底層細節，就可以完成非常穩定的分布式網絡通訊架構。

RSocket 作為通訊協議的后起之秀，核心是二進制異步化消息通訊，是否也能和 Spring Cloud 技術棧結合，實現服務注冊發現、客戶端負載均衡，從而更高效地實現面向服務的架構？這篇文章我們就討論一下 Spring Cloud 和 RSocket 結合實現服務注冊發現和負載均衡。

服務注冊發現

服務注冊發現的原理非常簡單，主要涉及三種角色：服務提供方、服務消費者和服務注冊中心。典型的架構如下：

服務提供方，如 RSocket Server，在應用啟動后，會向服務注冊中心注冊應用相關的信息，如應用名稱，ip 地址，Web Server 監聽端口號等，當然還會包括一些元信息，如服務的分組(group)，服務的版本號(version)，RSocket 的監聽端口號，如果是 WebSocket 通訊，還需要提供 ws 映射路徑等，不少開發者會將服務提供方的服務接口列表作為 tags 提交給服務注冊中心，方便后續的服務查詢和治理。

在本文中，我們采用 Consul 作為服務注冊中心，主要是 Consul 比較簡單，下載后執行 consul agent -dev 就可以啟動對應的服務，當然你可以使用 Docker Compose，配置也非常簡單，然后 docker-compose up -d 就可以啟動 Consul 服務。

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

假設你有一個服務應用，功能名稱為 calculator，同時提供兩個服務: 數學計算器服務(MathCalculatorService)和匯率計算器服務(ExchangeCalculatorService), ?那么我們該如何來命名該應用及其對應的服務接口名？

這里我們采用類似 Java package 命名規范，采用域名倒排的方式，如 calculator 應用對應的則為 com-example-calculator 樣式，為何是中劃線，而不是點？. 在 DNS 解析中作為主機名是非法的，只能作為子域名存在，不能作為主機名，而目前的服務注冊中心設計都遵循 DNS 規約，所以我們采用中劃線的方式來命名應用。這樣采用域名倒排和應用名結合的方式，可以確保應用之間不會重名，另外也方便和 Java Package 名稱進行轉換，也就是 -? 和 . 之間的相互轉換。

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

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

例如以下的服務命名都是合乎規范的：

• com.example.calculator.MathCalculatorService

• com.example.calculator.ExchangeCalculatorService

而 com.example.calculator.math.MathCalculatorService 則是錯誤的, ?因為在應用名稱和接口名稱之間多了 math。為何要采用這種命名規范？首先讓我們看一下服務消費方是如何調用遠程服務的。假設服務消費方拿到一個服務接口，如 com.example.calculator.MathCalculatorService，那么他該如何發起服務調用呢？

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

• 調用 ReactiveDiscoveryClient.getInstances(appName) 獲取應用名對應的服務實例列表(ServiceInstance)，ServiceInstance 對象會包含諸如 IP 地址，Web 端口號、RSocket 監聽端口號等其他元信息。

• 根據 RSocketRequester.Builder.transports(servers) 構建具有負載均衡能力的 RSocketRequester 對象。

• 使用服務全稱和具體功能名稱作為路由進行 RSocketRequester 的 API 調用，樣例代碼如下：

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

通過上述的命名規范，我們可以從服務接口全稱中提取出應用名，然后和服務注冊中心交互查找對應的實例列表，然后建立和服務提供者的連接，最后基于服務名稱進行服務調用。該命名規范，基本做到到了最小化的依賴，開發者完全是基于服務接口調用，非常簡單。

RSocket 服務編寫

有了服務的命名規范和服務注冊，編寫 RSocket 服務，這個還是非常簡單，和編寫一個 Spring Bean 沒有任何區別。引入 spring-boot-starter-rsocket 依賴，創建一個 Controller 類，添加對應的 MessagMapping annotation 作為基礎路由，然后實現功能接口添加功能名稱，樣例代碼如下：

@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); ? ? } }

上述代碼看起來好像有點奇怪，既然是服務實現，添加 @Controller 和 @MessageMapping，看起來好像有點不倫不類的。當然這些 annotation 都是一些技術細節體現，你也能看出，RSocket 的服務實現是基于 Spring Message 的，是面向消息化的。這里我們其實只需要添加一個自定義的 @SpringRSocketService annotation 就可以解決這個問題，代碼如下：

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

回到服務對應的實現代碼，我們改為使用 @SpringRSocketService annotation，這樣我們的代碼就和標準的 RPC 服務接口完全一模一樣啦，也便于理解。此外 @SpringRSocketService 和 @RSocketHandler 這兩個 Annotation，也方便我們后續做一些 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 依賴，設置一下 bootstrap.properties，然后在 application.properties 設置一下 RSocket 監聽的端口和元信息，我們還將該應用提供的服務接口列表作為 tags 傳給服務注冊中心，當然這個也是方便我們后續的服務管理。樣例如下：

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 服務應用啟動后，我們在 Consul 控制臺就可以看到服務注冊上來的信息，截屏如下：

RSocket 客戶端接入

客戶端接入稍微有一點復雜，主要是要基于服務接口全面要做一系列相關的操作，但是前面我們已經有了命名規范，所以問題也不大。客戶端應用同樣會接入服務注冊中心，這樣我們就可以獲得 ReactiveDiscoveryClient bean，接下來就是根據服務接口全名，如 com.example.calculator.ExchangeCalculatorService 構建出具有負載均衡的 RSocketRequester。

原理也非常簡單，前面說過，根據服務接口全稱，獲得其對應的應用名稱，然后調用 ReactiveDiscoveryClient.getInstances(appName) 獲得服務應用對應的實例列表，接下來將服務實例(ServiceInstance）列表轉換為 RSockt 的 LoadbalanceTarget 列表，其實就是 POJO 轉換，最后將轉 LoadbalanceTarget 列表進行 Flux 封裝(如使用 Sink 接口)，傳遞給 RSocketRequester.Builder 就完成具有負載均衡能力的 RSocketRequester 構建，詳細的代碼細節大家可以參考項目的代碼庫。

這里要注意的是接下來如何感知服務端實例列表的變化，如應用上下線，服務暫停等。這里我采用一個定時任務方案，定時查詢服務對應的地址列表。當然還有其他的機制，如果是標準的 Spring Cloud 服務發現接口，目前是需要客戶端輪詢的，當然也可以結合 Spring Cloud Bus 或者消息中間件，實現服務端列表變化的監聽。如果客戶端感知到服務列表的變化，只需要調用 Reactor 的 Sink 接口發送新的列表即可，RSocket Load Balance 在感知到變化后，會自動做出響應，如關閉即將失效的連接、創建新的連接等工作。

在實際的應用之間的相互通訊，會存在一些服務提供方不可用的情況，如服務方突然宕機或者其網絡不可用，這就導致了服務應用列表中部分服務不可用，那么 RSocket 這個時候會如何處理？不用擔心，RSocket Load Balance 有重試機制，當一個服務調用出現連接等異常，會重新從列表中獲取一個連接進行通訊，而那個錯誤的連接也會標識為可用性為 0，不會再被后續請求所使用。服務列表推送和通訊期間的容錯重試機制，這兩者保證了分布式通訊的高可用性。

最后讓我們啟動 client-app，然后從客戶端發起一個遠程的 RSocket 調用，截屏如下：

上圖中 com-example-calculator 服務應用包括三個實例，服務的調用會在這三個服務實例交替進行(RoundRobin 策略)。

開發體驗的一些考量

雖然服務注冊和發現、客戶端的負載均衡這些都完成啦，調用和容錯這些都沒有問題，但是還有一些使用體驗上的問題，這里我們也闡述一下，讓開發體驗做的更好。

1. 基于服務接口通訊

大多數 RPC 通訊都是基于接口的，如 Apache Dubbo、gRPC 等。那么 RSocket 能否做到？答案是其實完全可以。在服務端，我們已經是基于服務接口來實現 RSocket 服務啦，接下來我們只需要在客戶端實現基于該接口的調用就可以。對于 Java 開發者來說，這不是大問題，我們只需要基于 Java Proxy 機制構建就可以，而 Proxy 對應的 InvocationHandler 會使用 RSocketRequester 來實現 invoke() 的函數調用。詳細的細節請參考應用代碼中的的 RSocketRemoteServiceBuilder.java 文件，而且在 client-app module 中也已經包含了解基于接口調用的 bean 實現。

2. 服務接口函數的單參數問題

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

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

通過 interface 的 default method，我們可以為調用方提供給便捷函數，如在網絡傳輸的是字節數組 (byte[])，但是在 default 函數中，我們可以添加 File 對象支持，方便調用方使用。Interface 中的函數 API 負責服務通訊規約，default 函數來提升使用方的體驗，這兩者的配合，可以非常容易解決函數多參問題，當然 default 函數在一定程度上還可以作為數據驗證的前哨來使用。

3. RSocket Broker 支持

前面我們說到，RSocket 還有一種 Broker 架構，也就是服務提供方是隱藏在 Broker 之后的，請求主要是由 Broker 承接，然后再轉發給服務提供方處理，架構樣例如下：

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

回到 Broker 互通的場景，我們會向服務注冊中心查詢 broker0 對應的服務列表，然后和 broker0 集群的實例列表創建連接，這樣后續基于該接口的服務調用就會發送給 Broker 進行處理，也就是完成了服務注冊發現和 Broker 模式的混合使用的模式。

借助于這種定向指定服務接口和應用間的關聯，也方便我們做一些 beta 測試，如你想將 com.example.calculator.MathCalculatorService 的調用導流到 beta 應用，你就可以使用 com-example-calculator-beta1:com.example.calculator.MathCalculatorService 這種方式調用服務，這樣服務調用對應的流量就會轉發給 com-example-calculator-beta1 對應的實例，起到 beta 測試的效果。

回到最前面說到的規范，如果應用名和服務接口的綁定關系你實在做不到，那么你可以使用這種方式實現服務調用，如 calculator-server:com.example.calculator.math.MathCalculatorService，只是你需要更完整的文檔說明，當然這種方式也可以解決之前系統接入到目前的架構上，應用的遷移成本也比較小。如果你之前的面向服務化架構設計也是基于 interface 接口通訊的，那么通過該方式遷移到 RSocket 上完全沒有問題，對客戶端代碼調整也最小。

總結

通過整合服務注冊發現，結合一個實際的命名規范，就完成了服務注冊發現和 RSocket 路由之間的優雅配合，當然負載均衡也是包含其中啦。對比其他的 RPC 方案，你不需要引入 RPC 自己的服務注冊中心，復用 Spring Cloud 的服務注冊中心就可以，如 Alibaba Nacos， Consul， Eureka 和 ZooKeeper 等，沒有多余的開銷和維護成本。如果你想更多了解 RSocket RPC 相關的細節，可以參考 Spring 官方博客?《Easy RPC with RSocket》。

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

更多詳細的代碼細節，可以點擊鏈接查看文章對應的代碼庫！

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。