Spring RSocket：基於服務註冊發現的 RSocket 負載均衡

做者 | 雷卷
來源|阿里巴巴雲原生公衆號

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》。

更多詳細的代碼細節，能夠點擊連接查看文章對應的代碼庫！