Spring官方的RSocket Broker其實開發已經非常久了，我以為會伴隨著Spring Cloud 2021.0發布的，但是沒有發生。不過Spring RSocket Broker還是發布了最新的0.3版本，雖然還是預覽版，但目前已經可用，考慮官方還沒有提供對應的文檔，大家入門做Demo還有些困難，所以這篇文章就是幫你快速入門Spring RSocket Broker，同時解析一下RSocket Broker的特性。
一 Spring RSocket Broker架構
首先讓我們看一下Spring RSocket Broker的架構圖，如下：
RSocket Broker為一個集群對外提供服務，其主要服務就是應用注冊和RSocket請求的轉發，集群中的每一個Broker都維護著統一的全局路由表。RSocket Broker有兩個監聽端口：8001端口主要負責提供對外RSocket服務，如應用到Broker之間的長連接，然后就是該長連接之上的RSocket請求的發送和接收。7001端口主要負責集群內部Broker節點之間的通訊，如同步應用接入的元數據信息，確保全局服務路由表的統一，還包括Broker之間的請求轉發，當然Broker之間的通訊協議還是RSocket。
當一個服務應用和Broker建立連接時，會將一些基礎信息發送給Broker，對應的屬性主要包括：路由節點ID(routeId)、服務名稱(sevice-name)， tags(標簽)。
- 路由節點ID(routeId): 這個是應用到broker創建的長連接的唯一標識，通常是UUID，當然也可以用戶自己指定，連接的唯一性會讓各個Broker集群的全局路由表處理方便很多。注意routeId是長連接的標識，不是應用的標識，應用的標識是通過tags進行標識的。如果一個應用和broker創建兩條長連接，那么就有兩個不同的routeId，當然這種情況下每一條連接可以提供不同的服務名稱。
- 服務名稱：這個其實就是服務的DNS Name，如果其他應用想調用該應用提供的RSocket服務，就需要指定該服務名稱。雖然Spring RSocket提供了messageMapping，但是mapping的key是應用內部的，并不能保證全局唯一，只有Service Name + RSocket Mapping Key才能定位指定的服務，這個和http的原理是一致：Mapping Key類似HTTP Path，而Service Name類似域名。
- 標簽：標簽是用于標識應用提供的RSocket服務，當然RSocket Broker規范也提供了一些缺省的標簽，如InstanceName, ClusterName, Region, MajorVersion等。標簽不只是服務的元信息，此外還可以參與到服務路由上。如可以設置服務版本、集群名稱等，之前開發中老大難的定向路由就可以通過標簽輕松解決，如可以給服務打上InstanceName=app-leijuan的標簽，然后在調用中設置InsanceName就可以調用某位開發者啟動的服務實例。大家不用擔心基于Tag的路由性能問題，背后有Roaring BitMap支持，速度非常快。
應用可以向集群中的一個或者幾個RSocket Broker節點注冊，這個取決于Broker Client的配置，這個稍后我們還會講到。
注意: 這里大家不要將ServiceName僅理解為Java的Interface的全稱，如com.example.user.UserService，那么當一個應用有多個這樣的Java服務時，那么處理起來就比較麻煩啦。事實上serviceName主要用在請求路由上，如一個服務應用同時包括UerService, UserExtraService多個服務接口，你可以將ServiceName設置為com.example.user方式，當然還要保證ServiceName唯一，原先的RSocket routing key調整為UserService.findUserById這種Interface name + Method name方式，這樣就沒有問題啦。
當一個應用注冊到Broker上后，如果該應用想調用某一RSocket服務，只需要根據Service Name + Routing key就可以向Broker發起RSocket調用請求，然后Broker會根據內部的全局路由表，找到能夠提供服務的服務節點(RouteId)，然后將請求轉發給對應的服務節點，服務節點處理完畢后將response返回給Broker，Broker再將response返回給調用方。
如果當前Broker節點上并沒有對應的服務路由接入，這個是Broker會將請求轉發給有服務節點的Broker，這個就是請求轉發，然后再將那個Broker處理的結果返回給調用方。有同學可能會問，這里可能存在一個調用鏈死循環的問題，如broker1將請求轉發發給broker3，broker3上的服務突然下線也不存在，Broker3可能發回給Broker1，然后broker1再找其他broker發送?這種情況是不會發生的，Broker之間會同步應用上下線信息，所以每一個Broker都維護著集群統一的全局路由表，所以broker1給broker2轉發消息，broker2上一定有對應服務的route連接，即便broker2上突發狀況，服務對應的route沒啦，那么會再轉發給有服務的broker，當然如果都沒有找到，這個時候請求會被Broker保留(hold)住，在超時后會返回錯誤。
二 Spring RSocket Broker項目樣例
接下來我們就看一個真實的開發樣例，三個應用：一個Broker Server，一個服務提供者，一個服務調用者。Spring RSocket Broker對應的開發包已經提交到Maven倉庫，大家可以在文末鏈接查看。
1 RSocket Broker Server
RSocket Broker Server是一個標準的Spring Boot應用，你只需要在Spring Boot應用添加以下依賴：
- <dependency>
- <groupId>io.rsocket.broker</groupId>
- <artifactId>rsocket-broker-spring</artifactId>
- <version>0.3.0</version>
- </dependency>
然后在application.yaml文件中添加以下配置：
- io.rsocket.broker:
- uri: tcp://0.0.0.0:8001
- cluster.uri: tcp://0.0.0.0:7001
然后啟動Spring Boot應用，Broker也就啟動啦，并監聽7001和8001端口。有同學可能會問，為何不提供一個獨立的應用來啟動RSocket Broker?這個可能是Spring Cloud項目的出發點相關，和Spring Config Server，Registry Server一樣，都是被應用嵌入的，主要是方便開發者定制Broker的功能，如添加Web Console，對接Ops系統等，靈活性會就非常高。
2 RSocket Service Provider
接下來我們再創建一個Spring Boot應用，對外提供RSocket服務，首先添加一下以下依賴：spring-boot-starter-rsocket是標準的，方便Spring Boot應用集成RSocket，另外就是rsocket-broker-client-spring，這個是Broker Spring Client，負責完成和RSocket Broker的對接。
- <dependency>
- <groupId>org.springframework.boot</groupId>
- <artifactId>spring-boot-starter-rsocket</artifactId>
- </dependency>
- <dependency>
- <groupId>io.rsocket.broker</groupId>
- <artifactId>rsocket-broker-client-spring</artifactId>
- <version>0.3.0</version>
- </dependency>
然后在服務應用的application.yaml添加以下Broker Client配置，這里要說明service-name，建議采用前面談到的DNS命名方式，這樣可以確保服務名不會沖突。接下來就是brokers的地址列表，如下：
- io.rsocket.broker.client:
- service-name: com.example.PongService
- brokers:
- - tcp://localhost:8001
接下來我們還需要寫一個RSocket服務，這個就是標準的Spring RSocket，代碼如下：
- @Controller
- public class PongController {
- @MessageMapping("pong")
- public Mono<String> pong(String ping) {
- return Mono.just("hello " + ping);
- }
- }
然后我們啟動該服務應用，就會在RSocket Broker的日志輸出中看到該應用注冊到Broker的信息，這樣RSocket服務就完成了在Broker上的注冊。
3 RSocket Service Consumer
接下來我們還要創建一個應用用于調用RSocket服務，和服務應用一樣，添加相同的依賴，由于該應用并不對外提供RSocket服務，你將service-name調整為Namespace + 應用名稱即可，主要是不要和其他應用不要重名即可，如下：
- io.rsocket.broker.client:
- service-name: com.example.apps.${spring.application.name}
- brokers:
- - tcp://localhost:8001
接下來就是編寫一個Web Controller訪問RSocket服務，只需要注入BrokerRSocketRequester Bean，然后調用RSocket服務，這個和Spring RSocket的RSocketRequester使用方法類似，代碼如下：
- @RestController
- public class PortalController {
- @Autowired
- private BrokerRSocketRequester requester;
- @GetMapping("/")
- public Mono<String> index() {
- return requester.route("pong")
- .address("com.example.PongService")
- .data("ping")
- .retrieveMono(String.class);
- }
- }
啟動該應用后，你就可以使用curl命令進行測試，就可以看到熟悉的`Hello ping`輸出。
你有可能覺得這個客戶端調用比較原始， 其實你只要集成一下spring-retrosocket，然后就是你熟悉的Java接口，樣例如下：
- @RSocketBrokerClient
- interface GreetingClient {
- @MessageMapping("pong")
- @ServiceAddress("com.example.PongService")
- Mono<String> pong(String ping);
- }
三 Spring RSocket Broker的一些思考
1 RSocket Broker特性
Spring RSocket Broker開發已經挺久了，開發者都是Spring Cloud團隊成員，Oleh在Reactive和RSocket方面非常資深，Spencer也是Spring Cloud的核心架構師。Spencer在多個大會場合講述RSocket給Spring Cloud帶來的變化，完全是顛覆性的。從上述的應用樣例你也可以看出，不提Reactive全異步的性能，你不再需要服務注冊，你也不需要本地啟動接聽端口，介入Broker轉發后混各種云的服務都可以通過Broker進行相互調用。關于RSocket Broker的優點，Spring RSocket Broker有對應的說明，如下：
Routing and forwarding are used to forward RSocket requests between two RSocket connections via broker. In some cases, point-to-point interactions between a client and server are enough, in an enterprise environment, it is useful to decouple the client and server from each other. Some examples of why decoupling is necessary include blue/green deployments, load balancing, A/B testing, feature toggles, etc. Additionally, providing an intermediary can help with security and scalability. Finally, with the load balancing, routing and QoS, better overall application latency and throughput can be achieved than by direct connections.
2 RSocket Broker中直接通訊的解決方案
有同學可能會有疑問，經過RSocket Broker會有一定的性能損失，我這個應用QPS非常高，不能有延遲啊。這樣也沒有關系，還記得服務應用中添加了 spring-boot-starter-rsocket依賴嗎?我們只需要在application.yaml中添加以下配置項就可以打開RSocket的服務監聽端口，然后再通過RSocket Broker提供的元信息，然后我們使用RSocketRequester創建到目標服務的連接即可。有一些工作量，但是已經非常小，我們只需要使用reactor-pool自動管理直接連接的連接池就可以。
- spring.rsocket.server.transport: tcp
- spring.rsocket.server.port: 42252
3 RSocket Broker請求等待
Spring RSocket Broker還有一個特性就是服務上線延遲支持。舉一個例子，假設App-1和Service-1都在線上運行正常，突然Service-1的實例都下線啦，這個時候從app-1發出去的請求就找不到目標節點，這個時候該如何處理?
- 拒絕請求，馬上返回失敗：這個就是我們常說的快速失敗的設計，在企業架構設計中使用的比較多。如果是采用同步通訊和Thread Pool模式，你基本上必須使用該設計，不然較長時間的線程堵塞馬上讓你服務無法響應請求。
- 等待服務上線：就是Broker先保留(hold)請求，然后等服務上線后再轉發給上線的服務。這個設計有時非常有用，如在FaaS場景，Gateway上已經驗證該函數是存在的，目前只是函數下線，所以觸發一下函數上線然后請求等待就可以。此外還有就是網絡抖動的問題，會導致連接被中斷，這個時候也可以選擇等待服務上線。另外還有一個場景就是服務發布，如一個長尾服務只有一個實例，只要你能在3-5秒內將應用重新啟動完畢，這個時候broke就可以幫你hold住請求，再配合上客戶端的retry，即便只有一個實例，也不會感覺到服務被中斷。如在K8S中，將應用的鏡像設置為last，然后在設置為獲取最新，這樣一個redploy button就可以啦。當然這個請求等待時間也不是無限長的，你可以設置一個超時時間，然后返回錯誤就可以。
回到上述的場景，Broker-1這時候就會保留(hold)住請求，當service-1上線后，請求會馬上被轉發到上線的節點上處理。和同步通訊不一樣，異步的等待對Reactive系統并無系統壓力，所以等待服務上線完全是沒有問題。當然至于那種模式，你可以根據實際的技術需要進行選擇，RSocket Broker同時支持這兩個模式。
4 消息廣播模型
Spring RSocket Broker還支持消息的廣播，也就是將RSocket請求轉發給一批服務節點。消息廣播主要包括以下模型：
- fireAndForget模型：如配置推送場景，將配置更新請求發給service-name對應的服務列表即可。
- requestResponse模型：將請求發送給多個服務，然后將第一個響應的response返回給調用方，不論是響應的結果是成功還是失敗，這個和JavaScript的Promise.race()類似。Promise.race()特性，業務上場景好像不多，為了方便理解這里添加一個timeout，將其轉換為："在指定時間內最快地將結正確的結果返回"。如在數據同步的場景中，數據會同步到多個備份服務器上，但是由于種種原因，可能導致備份服務器上的數據同步出現延遲或者丟失，于是我向多個備份服務器查詢數據，并約定如果備份服務器上沒有該數據，則在1秒超時后返回異常，這樣就可以保證以最快的速度從備份服務器集群上拿到正確數據。當然在實際的架構中，我們會加上一層cache支持，避免同時向服務發起多個請求，浪費的資源也比較多。
- requestStream模型：將請求發送給多個服務方，然后將返回的stream進行匯總，然后將合并的消息流返回給調用方。在監控的場景非常有用，你希望各個服務或應用將5分鐘內的日志都匯報上來，然后進行統計，這個時候就非常有幫助。
- Channel模型：連續地向多個服務方的channel發送信息，然后將發出的信息再進行匯總。如你有多個指定要發送出去，然后將各個應用上對指定的響應結果進行匯總，就可以采用這個模型。
從上述的幾個模型看下來，基本上可以涵蓋眾多的廣播需求。如公司要你做一個config server進行配置推送，借助RSocket Broker是不是分分鐘就能搞定。借助RSocket Broker + Agent完成運維操作、日志采集等，是不是也不麻煩啦。在Prometheus的metrics定時采集場景，只需要發一個指令，就可以收集到所有機器上的metrics，比起向一臺臺節點發起HTTP請求，這種方式簡單很多。
5 Gateway和Broker
大多數的Gateway設計都采用主動連接的方式，也就是Gateway主動去連接上游服務的Proxy模式，當然要連接到上游服務還需要借助服務注冊發現，智能DNS等，其中的原理就不贅述啦。而Broker架構則采用被動模式，也就是等待服務連接到Broker上，也就是當服務Ready后，主動連接到Broker就可以，然后基于應用和Broker之間建立的長連接，進行請求轉發即可。對比Gateway架構，Broker模式簡單很多，內部不用管理上游服務的連接池，不需要服務注冊發現，當然對網絡也沒有特殊的打通要求，混合云的場景也適用等。
RSocket Broker雖然是基于RSocket協議的，但是還可以通過Bridge橋接的方式支持各種協議，如RSocket HTTP Bridge就可以擴展支持HTTP接入。
6 嵌入式的RSocket Broker
回答前面的問題，RSocket Broker是被應用嵌入的，你需要添加對應的依賴和配置，然后啟動對應的應用，這個和Spring Config Server等都是類似的，主要是方便開發者擴展Broker對應的特性，和其他系統進行集成。結合前面介紹的RSocket Broker特性，我們通過嵌入RSocket Broker，馬上就可以實現一些典型的業務場景：
- Config/Registry Server: 既然應用已經和Broker建立了長連接，元信息也都發送給Broker，所以Registry Server就水到渠成。RSocket Broker支持各種消息廣播模型，所以Config Server基本也就緒啦。單個應用的配置推送，基于獨立tag推送，基于Service Name整體推送，全部沒有問題。
- Web控制臺：嵌入Broker后，再開發一個web控制臺，這個對Spring Boot來說非常簡單。
- Data Gateway：如果你想做一個Data Gateway對外提供數據訪問服務，所有data worker節點連接到Broker，然后broker對外提供服務即可。
- Ops系統整合：這個使用Spring Boot整合即可，其他諸如對接入應用的健康度檢查等，這個只要發一個消息給應用即可。
- 應用和Broker的優雅上下線：通過推送brokers的配置信息，應用可以連接到新的brokers節點上，完成brokers集群的上下線。應用的上下線，在Broker集群中發一個ROUTE_REMOVE的消息即可，然后應用在3-5后即可下線。
7 Spring RSocket Broker Client
目前RSocket Broker的Client SDK主要包括Java和Node.js，但是其他語言的Broker Client SDK大家也不用擔心，Broker Client只是在RSocket SDK的Composite Metadata上添加一個新的 message/x.rsocket.broker.frame.v0 Metadata規范，借助于RSocket多語言SDK，主流語言開發的應用都可以快速接入到Broker上。
四 總結
最后有同學問道RSocket現在成熟了嗎?在Spring生態中，已經非常成熟。RSocket Java SDK由Spring團隊開發，Spring RSocket提供了RSocket和Spring的集成，Spring Boot內置rsocket-starter，Spring Cloud Function也添加了RSocket支持，考慮Java開發人員的習慣，還提供spring-retrosocket。其他Spring產品基本都支持了Reactive，所以對接通過Reactive就可以。可以RSocket外圍全部就緒啦，大家都在苦等RSocket Broker出現，這樣集成和部署就更簡單了。此外當前各種的各種服務，如REST API，GraphQL或者RPC框架，遷移到RSocket麻煩嗎?如果是Spring體現的，就是添加一個@MessageMapping的事情，然后就可以通過RSocket訪問這些服務 。Dubbo/HSF的服務，在Interface添加上spring-retrosocket提供的Annotation，然后就可以通過RSocket協議訪問這些服務啦。REST API在Controller基礎上添加@MessageMapping就可以。至于GraphQL，不需要任何調整，添加一個新的GraphqlController對接GraphQL底層服務即可。就目前的Spring RSocket Broker特性來說，對于一個中型企業，可以說不用什么調整，完全可以勝任，這個就是Spring Config Server，Spring Registry Server和Spring Cloud Gateway的定位是一樣的。
至于Spring Cloud團隊一直在講述RSocket對Spring Cloud和開發體驗的影響，相信閱讀了這里，你有了自己的判斷。但是還是那句老話：“世有伯樂，然后有千里馬。千里馬常有，而伯樂不常有”。
如果大家想了解更多的技術細節，我在樣例項目的的READM.md中進行了說明，有興趣可以訪問項目的Git倉庫。項目樣例的git倉庫地址為：https://github.com/linux-china/spring-rsocket-broker-demo 歡迎clone試用。
鏈接：https://repo1.maven.org/maven2/io/rsocket/broker/
原文鏈接：https://zhuanlan.51cto.com/art/202112/697463.htm
