作者 | 雷卷
來源|
阿裡巴巴雲原生公衆号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
com-example-calculator
- 調用
ReactiveDiscoveryClient.getInstances(appName)
- 根據
RSocketRequester.Builder.transports(servers)
- 使用服務全稱和具體功能名稱作為路由進行 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 釘釘群：
更多詳細的代碼細節，可以
點選連結檢視文章對應的代碼庫！