一個註解實現 WebSocket 集羣方案

介紹

WebSocket 大家應該是再熟悉不過了，如果是單體應用確實不會有什麼問題，但是當我們的項目使用微服務架構時，就可能會存在問題

比如服務 A 有兩個實例 A1 和 A2，前端的 WebSocket 客戶端 C 通過網關的負載均衡連到了 A1，這個時候當 A2 觸發消息發送的邏輯，需要將某個消息發送給所有的客戶端時，C 就接受不到消息

這個時候我們很快就能想到一種最簡單的解決方案，就是把 A2 的消息轉發給 A1，A1 再把消息發送給 C，這樣 C 就能收到 A2 發送的消息了

圖片

基於這個思路，我實現了一個庫，一個配置註解搞定一切

用法

接下來讓我們看看這個庫的用法

首先我們需要在啓動類上添加一個註解@EnableWebSocketLoadBalanceConcept

@EnableWebSocketLoadBalanceConcept
@EnableDiscoveryClient
@SpringBootApplication
public class AServiceApplication {

    public static void main(String[] args) {
        SpringApplication.run(AServiceApplication.class, args);
    }
}

接着我們在需要發送消息的地方注入WebSocketLoadBalanceConcept就可以愉快的跨實例發消息啦

@RestController
@RequestMapping("/ws")
public class WsController {

    @Autowired
    private WebSocketLoadBalanceConcept concept;

    @RequestMapping("/send")
    public void send(@RequestParam String msg) {
        concept.send(msg);
    }
}

是不是很簡單，有沒有覺得比自己集成單體應用的 WebSocket 還要簡單！

當你的同事還在頭疼要實現手動轉發時你已經通過一個配置註解實現了功能並開始泡茶喝

你的同事肯定對你刮目相看啊（又能開始摸魚了）

不知道大家看了之後是不是對具體實現已經有了一些思路呢

接下來我就來講講這個庫的實現流程

抽象思路

其實我之前有專門針對 WebSocket 實現過類似功能的模塊，只是當時的一些場景都是基於項目定死的，所以相對來說實現比較簡單，但是過於定製化不好擴展

有一天在和我的一個前同事聊天的過程中得知，他們在考慮讓設備和服務直連，並且服務要部署成多實例

設備和服務直連無非就是通過 TCP 這種長連接來實現，可以使用緩存來保存連接和服務地址的映射關係來實現點對點轉發的功能需求

聽到這裏，是不是感覺似曾相識？當時就有一道光穿過我的腦瓜子，真相只有一個！這不就和 WebSocket 在集羣模式下的問題一樣麼

於是我從原來針對 WebSocket 的思考，變成了對各種長連接的思考，最終我將這個問題抽象成了：長連接的集羣方案

而不管是 WebSocket 還是 TCP 都是長連接的一種具體實現

所以我們可以抽象一個頂級接口Connection，然後實現WebSocketConnection或者是TCPConnection

其實從抽象的角度來說不僅僅是長連接，短連接也在我們的抽象範圍之內，只不過類似 HTTP 等協議並不存在上述的問題，但是並不妨礙你實現一個HTTPConnection用於轉發消息，所以大家不要被先入爲主的思維束縛住了

轉發思路

之前講到，這個庫的主要思路就是將消息轉發給其他的服務實例來達到一個單播或廣播的效果

所以消息轉發的設計就非常重要了

首先消息轉發需要憑藉一些支持數據交互的技術手段

比如 HTTP，MQ，TCP，WebSocket

說到這裏。。。大家是不是。。。你 TM 原來自己就能搞定啊（掀桌）

長連接不就是用來交互數據的嗎，所以完全可以自給自足啊

於是就有一個精妙的想法在我腦子裏形成：

如果每個服務實例都把自己作爲一個客戶端，連接到其他服務上呢？

這樣其他服務實例就可以把消息發到這些僞裝的客戶端上，當服務實例上僞裝的客戶端接收到消息之後就可以再轉發給自己管理的真正的客戶端

撒花家人們，自閉（自我閉環）了屬於是

所以我們首先需要先讓服務實例之間相互連接上

連接流程

讓我們來看看互相建立連接是怎麼設計的

圖片

我定義了一個ConnectionSubscriber的接口，大家可以理解爲我們的服務實例要去訂閱監聽其他服務發送的消息

同時提供了默認實現，就是基於自身的協議進行連接和消息的發送

當然也能夠靈活的支持其他方式，只需要自定義一個ConnectionSubscriber就可以了，如果使用 MQ 的方式就可以實現一個MQConnectionSubscriber或者使用 HTTP 就可以實現一個HTTPConnectionSubscriber

只不過使用自身的協議就可以不用依賴其他的庫或是中間件了，當然如果你對消息的丟失率有比較嚴格的要求也可以使用 MQ 作爲消息轉發的中介，而以我之前參與過的項目來說，一般普通的 WebSocket 場景基本上還是能忍受一定的丟失率的

獲取服務實例信息

那麼我們怎麼知道要去連接哪些實例呢

我定義了一個ConnectionServerManager的接口用來管理服務信息

當然我們完全可以自己實現一個，比如通過配置文件來配置服務實例信息

不過我們有更方便的方式，那就是依賴 Spring Cloud 的服務發現組件了，不管是 Eureka 還是 Nacos 還是其他的註冊中心相當於都支持了，這就是抽象的魅力啊

我們可以通過DiscoveryClient#getInstances(Registration.getServiceId())來獲得所有的實例，排除掉自身就是需要連接的服務實例了

當我們的服務實例連接上其他的服務實例之後，發送一個自身實例信息的消息過去，其他的服務實例接收到對應的消息之後反過來連接我們的服務實例，保證一定的連接及時性，這樣雙方的連接就搭建起來了，可以互相轉發消息了

同時我還添加了心跳檢測和自動重連，當一段時間沒有收到心跳回復後就會斷開連接，並且每隔一段時間就會重新查詢一遍實例信息，如果發現存在某個服務實例沒有對應的連接，就會重新進行連接，這樣就能在某些偶爾網絡不好的情況下有一定的容錯

到目前爲止，我們基本的框架已經建立了，當我們啓動服務之後，服務間就會自動建立連接

連接區分和管理

基於上述的思路，我們肯定需要區分真實的客戶端和用來轉發的客戶端

於是我就把這些連接做了一個分類

圖片

然後對於這些連接進行一個統一的管理

圖片

通過連接工廠ConnectionFactory我們可以將任意的連接適配成Connection對象，並實現各種連接間的消息轉發

每個連接都會配置一個MessageEncoder和MessageDecoder用於消息的編碼和解碼，而且不同類別的連接對應的編碼器和解碼器肯定是不一樣的，比如轉發的消息和發給真實客戶端的消息很大程度上都是有區別的，所以額外定義了一個MessageCodecAdapter用來適配不同類型的編解碼器，也能讓大家在自定義時方便管理

消息發送

現在當我們發送某條消息之後，消息就會被轉發到其他的服務實例，所有的客戶端就都能收到了

不對啊，在有些情況下我們不想讓所有客戶端都收到啊，能不能我們想讓誰收到就讓誰收到啊

真麻煩，來，我把所有的連接都給你，你自己選吧

連接選擇

我們需要在消息發送時確定發送給哪些連接

圖片

於是我就定義了一個連接選擇器ConnectionSelector

每次要發送消息的時候，我都會匹配一個連接選擇器，然後通過選擇器來獲得需要發送消息的連接，而我們可以通過自定義連接選擇器來實現我們消息的精準發送

這裏其實就是我爲什麼會取名WebSocketLoadBalanceConcept的原因，爲什麼要叫 LoadBalance 呢

Ribbon 通過 IRule 來選擇一個 Server

我通過ConnectionSelector來選擇一個 Connection 集合

是不是有異曲同工之妙

繼續來說自定義選擇器

準備工作：

給指定用戶發送消息

很多場景下我們需要給指定的用戶發送消息

首先當客戶端連接上來時，可以通過參數或者主動發送一個消息將 userId 發給服務端，然後服務端將得到的 userId 存在Connection的metadata中

接着我們給需要發送的 Message 添加一個 header，將對應的 userId 作爲消息頭

這樣我們就可以自定義一個連接選擇器通過判斷 Message 是否包含 userId 消息頭來作爲匹配的條件，當 Message 的 headers 中存在 userId 時，對Connection中的 metadata 進行 userId 的匹配來篩選需要發送消息的連接

由於 userId 是唯一的，當我們自身服務連上來的客戶端中已經匹配到就不需要再轉發了，如果沒有匹配到就通過其他服務實例的客戶端進行消息轉發

庫中已經實現了對應的UserSelector和UserMessage，可以使用配置開啓並通過在連接路徑上添加 userId 參數來標記用戶

當然我們也可以借用緩存來精確的判斷需不需要轉發或者是需要轉發給哪幾個服務，把 userId 和服務的instanceId等一些具有唯一性的數據緩存在 Redis 中，當給用戶發送消息時，從 Redis 中獲得用戶對應的服務實例的instanceId或是具有唯一性的數據，如果經過匹配就是當前服務就可以直接下發，如果是其他服務就轉發給那個對應的服務就行了

給指定路徑發送消息

還有一種場景也比較常見就是類似主題訂閱，如訂閱設備狀態更新的數據，就要給每一個對應路徑的連接發送消息了

我們可以使用不同的路徑來表示不同主題，然後自定義一個連接選擇器來匹配連接的路徑和消息頭中指定的路徑

當然庫中也已經實現了對應的PathSelector和PathMessage，可以通過配置開啓

結束

最後請允許我發表一點對於抽象的拙見

抽象其實就和 “道生一，一生二，二生三，三生萬物” 一樣，根據你的頂級接口（也就是核心功能）不斷的向外展開，你的頂級接口就是道（狹義的來講）

以這個庫爲例，ConnectionLoadBalanceConcept就是這個庫的道，他的核心功能就是發送消息，至於怎麼發，發給誰，不確定，像是一個混沌的狀態

那麼什麼是一，二，三呢，我們發送消息需要載體於是就有了Connection和Message，我們需要對Connection進行管理於是就有了ConnectionRepository，我們需要轉發消息於是就有了ConnectionSubscriber等等

而萬物就像是具體的實現，是能落實的，基於 Spring Cloud 服務發現的連接管理器DiscoveryConnectionServerManager，基於路徑的連接選擇器PathSelector，基於 Reactive 的 WebSocket 連接ReactiveWebSocketConnection

就像是你創造的世界，不斷的衍生出各種各樣的規則，這些規則相輔相成，讓你的世界平穩的運行

當然你的世界也有可能存在 bug，手動狗頭

作者：代碼峽谷孫臏

來源：blog.csdn.net/m0_64360721/article/details/125125467

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/RLat1w8ZhLz0Y077AzPixw

猜你喜歡