端同步回調(diào)實(shí)現(xiàn))
這標(biāo)題看著長(zhǎng)其實(shí)拆開(kāi)就三層Spring Boot 里起一個(gè) WebSocket 服務(wù)端再內(nèi)嵌一個(gè) WebSocket 客戶(hù)端然后讓這個(gè)客戶(hù)端具備“發(fā)一條消息、同步等結(jié)果”的能力。別小看最后這點(diǎn)大部分網(wǎng)上的示例都卡在廣播和聊天室的階段真正拿進(jìn)業(yè)務(wù)系統(tǒng)里用的時(shí)候誰(shuí)都需要“我這端發(fā)出去一條指令對(duì)面到底處理成什么樣了”這種確定性的返回。這篇文章就把完整做法拆開(kāi)講透從依賴(lài)配置、端點(diǎn)編寫(xiě)、會(huì)話管理到客戶(hù)端封裝、同步回調(diào)協(xié)議、超時(shí)清理和重連保活全部給出可直接抄的代碼和參數(shù)建議。適合正在做即時(shí)通訊、設(shè)備指令下發(fā)或者想把長(zhǎng)連接能力接進(jìn) Spring Boot 業(yè)務(wù)代碼的朋友。1. 項(xiàng)目?jī)r(jià)值拆解為什么“同步回調(diào)”才是真正的業(yè)務(wù)剛需1.1 這個(gè)項(xiàng)目到底解決了什么問(wèn)題普通的 HTTP 接口天然就是“請(qǐng)求-響應(yīng)”模型你調(diào)一次接口拿到返回值事情就結(jié)束了。WebSocket 不一樣它是全雙工長(zhǎng)連接消息推出去之后對(duì)面的響應(yīng)什么時(shí)候回來(lái)、回沒(méi)回來(lái)都是異步事件。這就帶來(lái)一個(gè)很現(xiàn)實(shí)的矛盾Spring Boot 的業(yè)務(wù)代碼大部分是同步風(fēng)格Service 層調(diào)一個(gè)方法、拿到返回值、再往下走大家早就習(xí)慣了這種寫(xiě)法。如果 WebSocket 做成了“發(fā)消息 監(jiān)聽(tīng)回調(diào)”業(yè)務(wù)方就要被拆成兩段一段負(fù)責(zé)發(fā)一段負(fù)責(zé)在回調(diào)里繼續(xù)執(zhí)行代碼邏輯四分五裂排查問(wèn)題也很痛苦。標(biāo)題里的“客戶(hù)端消息同步回調(diào)”解決的就是這個(gè)矛盾對(duì)外暴露的接口仍然像普通方法一樣調(diào)用方傳參、等待、拿結(jié)果內(nèi)部則通過(guò) WebSocket 長(zhǎng)連接異步收發(fā)再靠請(qǐng)求 ID 把響應(yīng)和請(qǐng)求重新對(duì)齊。說(shuō)白了這是把異步通道“偽裝”回同步語(yǔ)義讓長(zhǎng)連接能力和常規(guī)業(yè)務(wù)代碼無(wú)縫銜接。我見(jiàn)過(guò)不少團(tuán)隊(duì)一開(kāi)始做長(zhǎng)連接模塊都是先寫(xiě)廣播 demo結(jié)果一接到真實(shí)業(yè)務(wù)里就發(fā)現(xiàn)各種不對(duì)勁核心原因就是沒(méi)想清楚“請(qǐng)求-響應(yīng)要對(duì)得上號(hào)”這件事。這個(gè)項(xiàng)目本質(zhì)上解決的不是連接問(wèn)題而是“確定性”問(wèn)題。1.2 三個(gè)典型業(yè)務(wù)場(chǎng)景我舉三個(gè)最常見(jiàn)的場(chǎng)景基本覆蓋了標(biāo)題里這套方案會(huì)被用到的位置。第一個(gè)是設(shè)備指令下發(fā)。管理平臺(tái)要控制一臺(tái)終端設(shè)備比如“開(kāi)啟攝像頭”“重啟服務(wù)”“調(diào)整參數(shù)”這種指令光發(fā)出去不算完你一定想知道設(shè)備端是否收到、是否執(zhí)行成功、如果失敗了錯(cuò)誤信息是什么。只有同步回調(diào)才能讓指令下發(fā)服務(wù)保持業(yè)務(wù)一致性不然指令狀態(tài)就只能靠輪詢(xún)數(shù)據(jù)庫(kù)去猜效率低而且實(shí)時(shí)性差。第二個(gè)是在線用戶(hù)操作回執(zhí)。IM 系統(tǒng)里不僅有聊天消息還有“拉黑”“撤回”“已讀回執(zhí)”這類(lèi)帶狀態(tài)的操作。服務(wù)端需要向指定的在線連接推送一條指令并等待用戶(hù)客戶(hù)端應(yīng)答。這種場(chǎng)景如果做成異步廣播你根本不知道被拉黑的那個(gè)用戶(hù)到底有沒(méi)有收到很多業(yè)務(wù)邏輯就沒(méi)法收斂。第三個(gè)是實(shí)時(shí)業(yè)務(wù)場(chǎng)景比如游戲?qū)纸Y(jié)算、交易確認(rèn)、協(xié)同編輯的版本同步。這些業(yè)務(wù)對(duì)結(jié)果返回的實(shí)時(shí)性要求極高HTTP 輪詢(xún)的延遲和開(kāi)銷(xiāo)都不合適WebSocket 長(zhǎng)連接加上同步回調(diào)能把原先“發(fā)了不管”的廣播式通信固化成一個(gè)帶結(jié)果的調(diào)用業(yè)務(wù)代碼寫(xiě)起來(lái)也自然很多?？梢哉f(shuō)只要你的系統(tǒng)里存在“在線設(shè)備或用戶(hù) 需要知道處理結(jié)果”的組合這套方案就是繞不開(kāi)的。1.3 標(biāo)題里的三層架構(gòu)每一層都缺一不可標(biāo)題雖然長(zhǎng)但核心架構(gòu)其實(shí)只有三層。服務(wù)端層負(fù)責(zé)連接管理、會(huì)話存儲(chǔ)、消息接收與推送解決的是“誰(shuí)能連、連上了往哪發(fā)”的問(wèn)題?？蛻?hù)端層是嵌入 Spring Boot 工程里的一個(gè) Java WebSocket 客戶(hù)端組件它對(duì)外暴露業(yè)務(wù)方法對(duì)內(nèi)維護(hù)長(zhǎng)連接狀態(tài)解決的是“業(yè)務(wù)代碼怎么調(diào)用”的問(wèn)題。同步回調(diào)層則是一套消息協(xié)議和等待機(jī)制通過(guò)請(qǐng)求 ID 把響應(yīng)消息和正在等待的請(qǐng)求重新配對(duì)解決的是“結(jié)果怎么回到調(diào)用方手里”的問(wèn)題。三層缺了任何一層都不完整沒(méi)有服務(wù)端客戶(hù)端沒(méi)東西可連沒(méi)有客戶(hù)端業(yè)務(wù)代碼就得裸寫(xiě)一堆 WebSocket 回調(diào)代碼難看還難維護(hù)沒(méi)有同步回調(diào)前面做的所有連接工作都停留在“能通”的水平業(yè)務(wù)根本沒(méi)法用起來(lái)。所以這篇博文會(huì)按這三層逐步拆開(kāi)寫(xiě)順序剛好就是標(biāo)題的閱讀順序。2. 技術(shù)選型與設(shè)計(jì)思路哪些方案能選、我為什么這樣選2.1 服務(wù)端集成 WebSocket 的三種主流姿勢(shì)Spring Boot 里做 WebSocket 服務(wù)端最常見(jiàn)的有三種路子基于ServerEndpoint注解、基于 Spring 的WebSocketHandler注冊(cè)、以及直接集成 Netty。我給它們拉了張對(duì)比表只看它們?cè)凇翱焖偌伞边@個(gè)場(chǎng)景下的表現(xiàn)方案上手難度代碼量生命周期處理適合場(chǎng)景ServerEndpoint注解低最少注解自帶 OnOpen/OnMessage/OnClose/OnError大多數(shù)業(yè)務(wù)項(xiàng)目快速接入沒(méi)問(wèn)題SpringWebSocketHandler中中等實(shí)現(xiàn)接口方法需要和 Spring MVC 深度整合、做攔截握手時(shí)集成 Netty高多自己管理通道和線程模型超大規(guī)模連接、TCP 層定制要求很高時(shí)標(biāo)題里明確寫(xiě)了“快速集成”那我的答案很直接選ServerEndpoint注解方案。原因不復(fù)雜一是它完全由 Java 生態(tài)規(guī)范定義和容器兼容性好代碼量最少二是每個(gè)生命周期回調(diào)都有明確的注解位置不需要去記 Spring 的接口簽名三是不需要額外引入路由邏輯路徑參數(shù)直接寫(xiě)在注解里比如/ws/{userId}在 OnOpen 方法里就能拿到用戶(hù) ID非常貼合業(yè)務(wù)需求。2.2 客戶(hù)端選型與依賴(lài)引入Java 生態(tài)里做 WebSocket 客戶(hù)端我一般直接用 Java-WebSocket 這個(gè)開(kāi)源庫(kù)。它提供了完整的客戶(hù)端實(shí)現(xiàn)連接、發(fā)送、監(jiān)聽(tīng)消息、心跳、重連這些能力都覆蓋了而且它是一個(gè)輕量庫(kù)不需要和 Spring 容器做任何綁定非常適合作為獨(dú)立組件封裝進(jìn)業(yè)務(wù)代碼里。相比之下用 Spring 自帶的WebSocketStompClient也不是不行但它強(qiáng)依賴(lài) STOMP 協(xié)議棧如果你的服務(wù)端是自己實(shí)現(xiàn)的普通 WebSocket 端點(diǎn)沒(méi)必要為握手過(guò)程引入這么重的依賴(lài)。Maven 坐標(biāo)很簡(jiǎn)單服務(wù)端只需要引入 Spring Boot 的 websocket starter客戶(hù)端則額外引入 Java-WebSocket 庫(kù)。注意版本方面選一個(gè)穩(wěn)定版本就行這十年下來(lái)它的 API 變化不大核心用法比較穩(wěn)定。在后面的代碼里我會(huì)把這兩段依賴(lài)分開(kāi)寫(xiě)清楚避免一些人把服務(wù)端和客戶(hù)端的依賴(lài)混在一起出現(xiàn)莫名奇妙的沖突。2.3 “同步回調(diào)”的本質(zhì)把異步消息拉回到同步時(shí)間線要說(shuō)清楚同步回調(diào)我打個(gè)比方。你去銀行柜臺(tái)辦事取了一個(gè)號(hào)碼牌然后坐下來(lái)等廣播叫號(hào)廣播喊到你的號(hào)你再上去拿結(jié)果。這里的號(hào)碼牌就是請(qǐng)求 ID廣播就是 WebSocket 的消息通知等待區(qū)就是你正在阻塞的業(yè)務(wù)線程。實(shí)現(xiàn)上就是用并發(fā)包里的CompletableFuture作為“等待憑證”每次發(fā)送請(qǐng)求前生成一個(gè)全局唯一的請(qǐng)求 ID同時(shí)創(chuàng)建一個(gè)CompletableFuture放進(jìn)一個(gè)ConcurrentHashMap里key 就是請(qǐng)求 ID。消息發(fā)出去之后調(diào)用方執(zhí)行future.get(timeout)阻塞等待。服務(wù)端處理完業(yè)務(wù)后回包時(shí)必須帶上同一個(gè)請(qǐng)求 ID客戶(hù)端在收到消息時(shí)用這個(gè) ID 去 Map 里找到對(duì)應(yīng)的 Future調(diào)用complete()把結(jié)果填進(jìn)去阻塞中的調(diào)用方立刻被喚醒拿到返回值繼續(xù)往下走。這就是“異步轉(zhuǎn)同步”的核心機(jī)制。需要注意這種同步是調(diào)用方線程層面的等待不是整個(gè)系統(tǒng)變成串行服務(wù)端依然是并發(fā)的只是每個(gè)請(qǐng)求在自己的線程里等待自己的結(jié)果彼此之間不干擾。同時(shí)future.get()一定要帶超時(shí)時(shí)間避免服務(wù)端一直不回包把業(yè)務(wù)線程全部打滿(mǎn)這一點(diǎn)在后面的代碼里也會(huì)處理掉。3. 服務(wù)端實(shí)現(xiàn)從握手建連到一對(duì)一消息推送3.1 依賴(lài)與基礎(chǔ)配置服務(wù)端的第一步是加依賴(lài)。Spring Boot 項(xiàng)目里引入spring-boot-starter-websocket就夠了它會(huì)把 WebSocket 相關(guān)的容器能力帶進(jìn)來(lái)同時(shí)和原有的 Web MVC 共存不會(huì)影響普通 HTTP 接口。dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-websocket/artifactId /dependency有了依賴(lài)之后還需要一個(gè)配置類(lèi)注冊(cè)ServerEndpointExporter。這個(gè)類(lèi)非常關(guān)鍵它的作用是掃描所有標(biāo)注了ServerEndpoint的類(lèi)把它們注冊(cè)到底層 WebSocket 容器里讓路徑真正生效。如果少了這一步你寫(xiě)的端點(diǎn)類(lèi)不會(huì)報(bào)錯(cuò)但連接時(shí)永遠(yuǎn)是連接失敗這個(gè)坑不少人都踩過(guò)。Configuration public class WebSocketConfig { Bean public ServerEndpointExporter serverEndpointExporter() { return new ServerEndpointExporter(); } }這一步完成后服務(wù)端的骨架就搭好了。后面寫(xiě)的端點(diǎn)類(lèi)、會(huì)話管理工具、消息中心都會(huì)被 Spring Boot 的自動(dòng)配置能力串起來(lái)。3.2 會(huì)話管理并發(fā)安全是第一個(gè)考點(diǎn)WebSocket 服務(wù)端的核心資源就是Session每個(gè)連接對(duì)應(yīng)一個(gè)會(huì)話我們得把“某個(gè)用戶(hù) ID ? 某個(gè) Session”的對(duì)應(yīng)關(guān)系管理好。因?yàn)?WebSocket 連接的建立和斷開(kāi)是隨時(shí)的可能幾十個(gè)線程同時(shí)在線增刪會(huì)話所以這里必須用并發(fā)安全的容器。我用一個(gè)獨(dú)立的工具類(lèi)來(lái)管理在線會(huì)話內(nèi)部使用ConcurrentHashMap以用戶(hù) ID 為 key以Session對(duì)象為 value。提供四個(gè)最基本的方法加會(huì)話、刪會(huì)話、查會(huì)話、統(tǒng)計(jì)在線數(shù)。為什么單獨(dú)抽一個(gè)類(lèi)因?yàn)槎它c(diǎn)類(lèi)和消息中心都會(huì)用到這份數(shù)據(jù)如果堆在前端類(lèi)里代碼會(huì)越來(lái)越亂還會(huì)出現(xiàn)靜態(tài)變量和實(shí)例變量混淆的問(wèn)題。public class SessionManager { private static final MapString, Session ONLINE_SESSIONS new ConcurrentHashMap(); public static void add(String userId, Session session) { ONLINE_SESSIONS.put(userId, session); } public static void remove(String userId) { ONLINE_SESSIONS.remove(userId); } public static Session get(String userId) { return ONLINE_SESSIONS.get(userId); } public static int onlineCount() { return ONLINE_SESSIONS.size(); } }這里要特別提醒一點(diǎn)同一個(gè)用戶(hù)如果重復(fù)連接比如瀏覽器刷新、客戶(hù)端斷線重連舊的 Session 沒(méi)有正常關(guān)閉Map 里就會(huì)殘留失效連接。所以在add的時(shí)候建議先判斷一下舊 Session 是否還存在如果存在就先關(guān)掉再放新的進(jìn)去。否則你后續(xù)推送時(shí)可能拿到一個(gè)已經(jīng)關(guān)閉的 Session白白增加很多isOpen()判斷。3.3 WebSocket 端點(diǎn)的完整實(shí)現(xiàn)端點(diǎn)類(lèi)直接寫(xiě)在 Spring Boot 工程里加Component和ServerEndpoint兩個(gè)注解。路徑我設(shè)計(jì)成/ws/{userId}這樣連接時(shí)天然帶上了身份信息服務(wù)端不用再通過(guò)消息內(nèi)容去猜是哪個(gè)用戶(hù)。Component ServerEndpoint(/ws/{userId}) public class ChatWebSocketEndpoint { OnOpen public void onOpen(Session session, PathParam(userId) String userId) { SessionManager.add(userId, session); System.out.println(連接建立: userId userId , sessionId session.getId()); } OnMessage public void onMessage(String message, Session session) { System.out.println(收到消息: message); // 這里交給業(yè)務(wù)邏輯處理注意線程池隔離 MessageRouter.route(session, message); } OnClose public void onClose(Session session, PathParam(userId) String userId) { SessionManager.remove(userId); System.out.println(連接關(guān)閉: userId userId); } OnError public void onError(Session session, Throwable error) { error.printStackTrace(); } }這段代碼看起來(lái)簡(jiǎn)單但有幾個(gè)細(xì)節(jié)值得說(shuō)。第一PathParam必須和路徑占位符里的名字一致不然后臺(tái)會(huì)一直打參數(shù)解析異常的日志連接成功率看起來(lái)沒(méi)問(wèn)題但用戶(hù) ID 永遠(yuǎn)是空值。第二onError里最好不要把異常只打日志就算了應(yīng)該把當(dāng)前會(huì)話刪掉避免一個(gè)半死不活的連接還掛在在線列表里占位置。第三onMessage收到的消息體格式建議直接約定為 JSON服務(wù)端不要接收裸字符串否則后面做回調(diào)配對(duì)時(shí)請(qǐng)求 ID 都沒(méi)地方放。3.4 消息中心設(shè)計(jì)對(duì)點(diǎn)推送與廣播有了一對(duì)一的 Session 映射下一步就是封裝一個(gè)消息中心讓業(yè)務(wù)代碼不用直接碰Session對(duì)象。推送消息的核心方法是sendToUser它負(fù)責(zé)根據(jù)用戶(hù) ID 找到 Session檢查連接狀態(tài)然后發(fā)送文本消息。public class MessageCenter { public static void sendToUser(String userId, String message) { Session session SessionManager.get(userId); if (session ! null session.isOpen()) { try { session.getBasicRemote().sendText(message); } catch (IOException e) { SessionManager.remove(userId); e.printStackTrace(); } } } public static void broadcast(String message) { for (String userId : SessionManager.allUserIds()) { sendToUser(userId, message); } } }這里我用了getBasicRemote().sendText()它是同步發(fā)送線程會(huì)等消息寫(xiě)完才返回。如果推送的頻率特別高建議改用getAsyncRemote().sendText()異步發(fā)送不用等 IO 完成但這意味著你要自己處理消息順序和發(fā)送失敗的回調(diào)復(fù)雜度會(huì)上去。另外一個(gè)很多人在推送時(shí)忽略的點(diǎn)同一個(gè)Session不要多個(gè)線程同時(shí)調(diào)用sendText()會(huì)出現(xiàn)消息交錯(cuò)甚至直接拋異常。并發(fā)量大的時(shí)候可以考慮給每個(gè) Session 加一個(gè)寫(xiě)鎖或者用隊(duì)列把所有推送任務(wù)排隊(duì)處理。我在實(shí)際排查過(guò)的問(wèn)題里見(jiàn)過(guò)好幾次“消息偶爾串線”的詭異現(xiàn)象最后定位就是并發(fā)寫(xiě)同一個(gè) Session 造成的。3.5 一個(gè)容易踩的坑Spring 容器與端點(diǎn)實(shí)例的隔離ServerEndpoint的端點(diǎn)類(lèi)和普通的Component有一個(gè)本質(zhì)區(qū)別端點(diǎn)實(shí)例是由 WebSocket 容器創(chuàng)建和管理的每次連接通常對(duì)應(yīng)新的實(shí)例它不是 Spring 的單例。很多人在這里栽過(guò)跟頭在端點(diǎn)類(lèi)里直接Autowired一個(gè) Service結(jié)果發(fā)現(xiàn)是 null因?yàn)?Spring 根本沒(méi)有管理那個(gè)端點(diǎn)實(shí)例。解決辦法是用靜態(tài)字段加 setter 注入的方式把 Spring 容器里的 Service 暴露到端點(diǎn)類(lèi)里。具體做法是端點(diǎn)類(lèi)加Component讓 Spring 掃描到然后在類(lèi)里定義一個(gè)靜態(tài)的 Service 字段和 setter 方法setter 上加Autowired。這樣 Spring 在初始化時(shí)會(huì)把 Service 注入到 setter 里再賦給靜態(tài)字段端點(diǎn)實(shí)例后續(xù)就可以直接使用這個(gè)靜態(tài)字段了。Component ServerEndpoint(/ws/{userId}) public class ChatWebSocketEndpoint { private static MessageService messageService; Autowired public void setMessageService(MessageService service) { ChatWebSocketEndpoint.messageService service; } OnMessage public void onMessage(String message, Session session) { Object result messageService.handle(message); // 繼續(xù)處理 result ... } }這是很多教程里不會(huì)提到的關(guān)鍵細(xì)節(jié)。如果你打電話報(bào)錯(cuò)找不到 Service十有八九就是這里沒(méi)處理好。3.6 服務(wù)端如何配合同步回調(diào)協(xié)議服務(wù)端本身不一定知道“同步回調(diào)”這回事它只需要遵守一條協(xié)議回包時(shí)必須帶上請(qǐng)求里攜帶的reqId。這個(gè)約定是整個(gè)同步機(jī)制成立的基礎(chǔ)。我建議在項(xiàng)目里提前定好統(tǒng)一的消息外殼格式。請(qǐng)求消息和響應(yīng)消息都包含三個(gè)基礎(chǔ)字段reqId表示請(qǐng)求 IDtype表示消息類(lèi)型payload放業(yè)務(wù)數(shù)據(jù)。這樣無(wú)論消息走的是同步請(qǐng)求還是異步事件接收方只要掃一眼type就能分流通過(guò)reqId就能配對(duì)。OnMessage public void onMessage(String message, Session session) { JSONObject request JSON.parseObject(message); String reqId request.getString(reqId); String payload request.getString(payload); Object result messageService.handle(payload); JSONObject response new JSONObject(); response.put(reqId, reqId); response.put(status, 200); response.put(data, result); MessageCenter.sendToUser(getUserId(session), response.toJSONString()); }服務(wù)端回包時(shí)只要找回請(qǐng)求里的reqId填進(jìn)響應(yīng)體客戶(hù)端的同步等待就能被正確喚醒。如果哪天發(fā)現(xiàn)客戶(hù)端一直超時(shí)第一個(gè)排查點(diǎn)就是服務(wù)端有沒(méi)有把reqId原樣返回。這個(gè)看起來(lái)不起眼卻是我見(jiàn)過(guò)最多的失誤原因之一。4. 客戶(hù)端封裝與同步回調(diào)落地4.1 客戶(hù)端核心骨架與連接管理這里的客戶(hù)端指的是嵌入 Spring Boot 工程里的 Java WebSocket 客戶(hù)端組件不是瀏覽器頁(yè)面。它會(huì)去連接遠(yuǎn)端的 WebSocket 服務(wù)然后把一個(gè)“同步請(qǐng)求方法”暴露給上層業(yè)務(wù)調(diào)用。依賴(lài)如下我選用 Java-WebSocket 庫(kù)來(lái)做客戶(hù)端dependency groupIdorg.java-websocket/groupId artifactIdJava-WebSocket/artifactId version1.5.4/version /dependency客戶(hù)端組件核心就是創(chuàng)建一個(gè)WebSocket實(shí)例綁定服務(wù)端地址并在onMessage回調(diào)里集中處理收到的消息。連接建立之后上層業(yè)務(wù)不需要關(guān)心連接對(duì)象只需要調(diào)用我們封裝的同步方法。public class WsClient { private WebSocket socket; public void connect(String serverUrl) throws Exception { socket new WebSocket(new URI(serverUrl)) { Override public void onMessage(String message) { handleResponse(message); } Override public void onClose(int code, String reason, boolean remote) { scheduleReconnect(serverUrl); } }; socket.connect(); // 等待連接完成 CountDownLatch latch new CountDownLatch(1); // 這里可做握手完成的標(biāo)志位簡(jiǎn)化示例直接按需等待 } }有一點(diǎn)要注意connect()是非阻塞的真正建立連接需要幾毫秒到幾十毫秒不等。所以如果你的業(yè)務(wù)方剛調(diào)用完connect()立刻就去發(fā)消息很可能會(huì)拋“連接未打開(kāi)”的異常。一個(gè)穩(wěn)妥的做法是在 onOpen 回調(diào)里放一個(gè)門(mén)閂等連接建立后再開(kāi)放發(fā)消息。我實(shí)際做的時(shí)候是在內(nèi)部維護(hù)了一個(gè)AtomicBoolean connected標(biāo)志每次發(fā)送前檢查沒(méi)連上就阻塞等待一會(huì)兒。4.2 請(qǐng)求-響應(yīng)對(duì)齊的關(guān)鍵結(jié)構(gòu)整個(gè)同步回調(diào)機(jī)制最關(guān)鍵的數(shù)據(jù)結(jié)構(gòu)有兩塊一個(gè)是請(qǐng)求 ID 生成器一個(gè)是等待響應(yīng)的 Future 容器。請(qǐng)求 ID 我用AtomicLong自增序列來(lái)生成配合一個(gè)簡(jiǎn)單的前綴保證在本進(jìn)程內(nèi)唯一。如果多個(gè)實(shí)例部署建議用 UUID 或者在前綴里加上實(shí)例標(biāo)識(shí)避免把不同實(shí)例的請(qǐng)求響應(yīng)匹配錯(cuò)。public class WsClient { private static final AtomicLong REQ_SEQ new AtomicLong(0); private final MapString, CompletableFutureString pendingMap new ConcurrentHashMap(); private String generateReqId() { return REQ_ REQ_SEQ.incrementAndGet(); } }pendingMap就是那個(gè)“等待區(qū)”。每個(gè)請(qǐng)求發(fā)出前往 Map 里放一個(gè)CompletableFuture收到對(duì)應(yīng)響應(yīng)時(shí)從這個(gè) Map 里取出 Future把它完成掉。Map 用ConcurrentHashMap是因?yàn)榭赡苡卸鄠€(gè)業(yè)務(wù)線程同時(shí)發(fā)起請(qǐng)求、多個(gè)消息回調(diào)線程同時(shí)處理響應(yīng)必須在并發(fā)場(chǎng)景下保證安全。4.3 封裝同步請(qǐng)求方法核心方法sendSync的邏輯這樣設(shè)計(jì)生成請(qǐng)求 ID、封裝消息、注冊(cè) Future、發(fā)送消息、阻塞等待、最后無(wú)論如何都要從 Map 里清理掉這個(gè) Future。清理這一步絕對(duì)不能省否則服務(wù)端一直不回包Map 里就會(huì)堆積大量的 Future 對(duì)象時(shí)間一長(zhǎng)就是內(nèi)存泄漏。public String sendSync(String payload, long timeout, TimeUnit unit) throws Exception { String reqId generateReqId(); CompletableFutureString future new CompletableFuture(); pendingMap.put(reqId, future); JSONObject request new JSONObject(); request.put(reqId, reqId); request.put(type, BIZ_REQUEST); request.put(payload, payload); socket.send(request.toJSONString()); try { return future.get(timeout, unit); } finally { pendingMap.remove(reqId); } }這個(gè)方法的調(diào)用方式非常直白業(yè)務(wù)方傳一段 JSON 字符串傳入超時(shí)時(shí)間然后同步拿到服務(wù)端返回的另一個(gè) JSON 字符串。如果服務(wù)端那邊處理時(shí)間大致可預(yù)估建議把超時(shí)設(shè)置成 3 到 5 秒如果包含復(fù)雜的業(yè)務(wù)鏈路可以放寬到 10 秒但超過(guò) 30 秒我一般不建議因?yàn)殚L(zhǎng)時(shí)間阻塞業(yè)務(wù)線程會(huì)讓整個(gè)應(yīng)用的線程池很快耗盡。4.4 心跳保活與斷線重連長(zhǎng)連接有個(gè)天然的敵人網(wǎng)絡(luò)中間設(shè)備和服務(wù)器本身會(huì)清理閑置連接。Tomcat 默認(rèn)會(huì)有空閑連接超時(shí)Nginx 代理也有讀寫(xiě)超時(shí)如果客戶(hù)端長(zhǎng)時(shí)間不發(fā)送任何數(shù)據(jù)連接就會(huì)被對(duì)端靜默關(guān)閉?？蛻?hù)端自己感知不到等到下次發(fā)消息才發(fā)現(xiàn)連接已經(jīng)斷了這時(shí)同步請(qǐng)求就直接卡到超時(shí)。所以客戶(hù)端必須加心跳。我一般用ScheduledExecutorService每隔 10 秒發(fā)送一個(gè)心跳消息消息體統(tǒng)一成{type:PING}。服務(wù)端收到 PING 時(shí)不需要回業(yè)務(wù)數(shù)據(jù)但可以回一個(gè) PONG或者只要服務(wù)端還在正常讀數(shù)據(jù)客戶(hù)端這條心跳就起到了?；钭饔?。private void startHeartbeat() { ScheduledExecutorService executor Executors.newSingleThreadScheduledExecutor(); executor.scheduleAtFixedRate(() - { if (socket ! null socket.isOpen()) { JSONObject ping new JSONObject(); ping.put(type, PING); socket.send(ping.toJSONString()); } }, 10, 10, TimeUnit.SECONDS); }斷線重連方面我會(huì)在onClose里觸發(fā)重連邏輯但重連不能做成無(wú)腦循環(huán)否則服務(wù)端還沒(méi)恢復(fù)客戶(hù)端就會(huì)以極高的頻率瘋狂創(chuàng)建連接把服務(wù)端打得更慘。建議做退避重試第一次 1 秒、第二次 2 秒、第三次 4 秒指數(shù)遞增最多間隔 60 秒服務(wù)端恢復(fù)連接后把重連次數(shù)清零。重連期間所有正在等待的sendSync請(qǐng)求會(huì)因?yàn)槌瑫r(shí)統(tǒng)一失敗上層業(yè)務(wù)要接受這種“短暫不可用”的現(xiàn)實(shí)并在失敗后走自己的重試策略。4.5 回調(diào)協(xié)議的最佳實(shí)踐與參數(shù)建議這部分我總結(jié)了幾個(gè)真正算得上經(jīng)驗(yàn)的東西。第一消息格式必須是 JSON并且在項(xiàng)目一開(kāi)始就定好外殼字段不能今天用reqId明天用requestId這種不一致會(huì)讓整個(gè)調(diào)試過(guò)程變成災(zāi)難。建議統(tǒng)一為reqId、type、payload三個(gè)字段同步響應(yīng)再加一個(gè)status表示結(jié)果狀態(tài)碼。第二sendSync返回的響應(yīng)建議不要直接返回原始字符串最好解析成JSONObject再返回同時(shí)把status非 200 的響應(yīng)當(dāng)成業(yè)務(wù)異常拋出去。這樣上層業(yè)務(wù)代碼拿到的是干凈的結(jié)果而不是還要自己判斷一堆狀態(tài)位。第三請(qǐng)求 ID 最好在客戶(hù)端生成。如果服務(wù)端也可以生成就會(huì)出現(xiàn)兩邊 ID 體系不一致的問(wèn)題調(diào)試時(shí)很容易看花眼?？蛻?hù)端生成后原樣傳給服務(wù)端服務(wù)端原樣回傳鏈路清晰問(wèn)題也好定位。第四同步回調(diào)適合“必須有結(jié)果”的操作但不是所有消息都適合同步。像心跳、狀態(tài)上報(bào)、位置推送這類(lèi)頻率高、不關(guān)心響應(yīng)的消息建議走獨(dú)立的異步通道不要通過(guò)sendSync發(fā)否則線程池重負(fù)載下這些消息也會(huì)排隊(duì)等待白白增加延遲。5. 實(shí)戰(zhàn)避坑與排查實(shí)錄5.1 常見(jiàn)問(wèn)題速查表我把實(shí)際開(kāi)發(fā)中遇到過(guò)的問(wèn)題整理成了一張表按現(xiàn)象排查效率很高現(xiàn)象可能原因解決辦法客戶(hù)端始終連接不上服務(wù)端缺少ServerEndpointExporter注冊(cè)檢查配置類(lèi)是否注冊(cè)了該 Bean連接報(bào) 400 握手失敗路徑參數(shù)不對(duì)或請(qǐng)求路徑與端點(diǎn)不匹配核對(duì)ServerEndpoint路徑和客戶(hù)端連接地址請(qǐng)求 100% 超時(shí)reqId沒(méi)有配對(duì)服務(wù)端響應(yīng)里丟了reqId抓包看服務(wù)端回包字段確認(rèn)reqId原樣返回連接頻繁斷開(kāi)沒(méi)有心跳中間設(shè)備回收空閑連接客戶(hù)端加心跳建議 10 秒一次消息出現(xiàn)錯(cuò)亂或串線同一 Session 被多個(gè)線程并發(fā)寫(xiě)加 Session 寫(xiě)鎖或全部改成異步發(fā)送重連后消息發(fā)不出去重連成功后內(nèi)部連接狀態(tài)沒(méi)更新在 onOpen 回調(diào)里更新連接標(biāo)志位長(zhǎng)時(shí)間運(yùn)行后內(nèi)存增長(zhǎng)明顯pendingMap中超時(shí)的 Future 沒(méi)清理finally塊里執(zhí)行pendingMap.remove(reqId)5.2 幾個(gè)我踩過(guò)的深坑第一個(gè)深坑就是忘記注冊(cè)ServerEndpointExporter。項(xiàng)目啟動(dòng)完全正常沒(méi)有任何報(bào)錯(cuò)但所有 WebSocket 連接都失敗我在瀏覽器控制臺(tái)看到的是 404。排查了半天才發(fā)現(xiàn)是配置類(lèi)里少了一個(gè) Bean這種問(wèn)題報(bào)錯(cuò)信息不直接非常迷惑人。第二個(gè)深坑是瀏覽器里連接 503。如果你前面掛了 Nginx 之類(lèi)的反向代理必須在代理配置里顯式開(kāi)啟協(xié)議升級(jí)并且設(shè)置合理的超時(shí)時(shí)間。很多人本地直連正常一到測(cè)試環(huán)境就失敗基本都是這一層沒(méi)配置。生產(chǎn)環(huán)境還要注意負(fù)載均衡器的會(huì)話保持不然同一個(gè)客戶(hù)端的多次連接會(huì)被轉(zhuǎn)發(fā)到不通的節(jié)點(diǎn)上。第三個(gè)深坑是普通方法里調(diào)用sendSync導(dǎo)致線程池打滿(mǎn)。我當(dāng)時(shí)在業(yè)務(wù)線程里直接調(diào)同步請(qǐng)求結(jié)果服務(wù)端回包變慢幾十個(gè)請(qǐng)求全部卡在future.get()上業(yè)務(wù)線程全部占滿(mǎn)應(yīng)用開(kāi)始出現(xiàn)連環(huán)超時(shí)。后來(lái)給同步請(qǐng)求單獨(dú)隔離了一個(gè)線程池并且嚴(yán)格限制超時(shí)時(shí)間這個(gè)問(wèn)題才算解決。第四個(gè)深坑是關(guān)于 Resend 或者消息重發(fā)。WebSocket 底層不保證消息不丟有時(shí)候服務(wù)端確實(shí)不會(huì)回包客戶(hù)端超時(shí)后會(huì)重試。重試時(shí)不能直接把同一個(gè)reqId重新發(fā)一遍否則如果服務(wù)端前面已經(jīng)處理成功只是回包丟了會(huì)引起業(yè)務(wù)重復(fù)執(zhí)行。建議給每條消息加一個(gè)sequence序號(hào)重試時(shí)用新reqId但在payload里帶上原始業(yè)務(wù)冪等號(hào)服務(wù)端自己決定是否要去重。5.3 分布式多實(shí)例下的拓展思路以上所有方案在單機(jī)模式下都是沒(méi)有問(wèn)題的因?yàn)闀?huì)話信息都存在本地 JVM 的ConcurrentHashMap里。一旦項(xiàng)目從單實(shí)例擴(kuò)展成多實(shí)例部署問(wèn)題就出來(lái)了用戶(hù) A 連接在實(shí)例 1用戶(hù) B 連接在實(shí)例 2實(shí)例 1 想推送消息給 B本地 Session 里根本找不到 B 的連接。常規(guī)解法是把“在線會(huì)話表”從本地內(nèi)存挪到外部存儲(chǔ)比如 Redis 哈希結(jié)構(gòu)里記錄 userId 和節(jié)點(diǎn) ID 的映射。推送時(shí)先查路由表找到目標(biāo)節(jié)點(diǎn)再把消息投遞到對(duì)應(yīng)的實(shí)例由該實(shí)例的本地會(huì)話表完成真正的推送。節(jié)點(diǎn)間通信可以用消息隊(duì)列或者 Redis 發(fā)布訂閱來(lái)做實(shí)例只需消費(fèi)消息并判斷消息里的目標(biāo)節(jié)點(diǎn)是不是自己。同步回調(diào)在這種架構(gòu)下也會(huì)變得復(fù)雜但好在請(qǐng)求 ID 的機(jī)制不受影響只要消息能路由到正確的節(jié)點(diǎn)配對(duì)邏輯依然成立。我通常建議先把單機(jī)版跑通再考慮加分布式路由層不要一上來(lái)就把復(fù)雜度拉滿(mǎn)。最后再分享一點(diǎn)個(gè)人體會(huì)整個(gè)項(xiàng)目跑通之后我最大的感受是同步回調(diào)這個(gè)能力真正難的不是代碼而是協(xié)議先想清楚。你只要把reqId、type、payload、status這套外殼定好后面換服務(wù)端框架、換客戶(hù)端引擎都不傷筋動(dòng)骨。我后來(lái)遇到類(lèi)似的需求第一件事永遠(yuǎn)是先確認(rèn)消息格式而不是先寫(xiě)連接代碼。另外一個(gè)很實(shí)際的心得是同步語(yǔ)義雖好但不要貪只對(duì)業(yè)務(wù)上確實(shí)需要返回結(jié)果的請(qǐng)求用同步像心跳、狀態(tài)廣播這些應(yīng)該走純事件否則你的線程池很快就會(huì)被占滿(mǎn)。要是后續(xù)打算跨集群擴(kuò)展可以把請(qǐng)求 ID 和路由信息塞進(jìn)消息頭再配一個(gè)消息網(wǎng)關(guān)這套客戶(hù)端組件基本不用改服務(wù)端的路由層單獨(dú)替換就行。 Spring Boot WebSocket客戶(hù)端同步回調(diào)實(shí)現(xiàn)
這標(biāo)題看著長(zhǎng)其實(shí)拆開(kāi)就三層Spring Boot 里起一個(gè) WebSocket 服務(wù)端再內(nèi)嵌一個(gè) WebSocket 客戶(hù)端然后讓這個(gè)客戶(hù)端具備“發(fā)一條消息、同步等結(jié)果”的能力。別小看最后這點(diǎn)大部分網(wǎng)上的示例都卡在廣播和聊天室的階段真正拿進(jìn)業(yè)務(wù)系統(tǒng)里用的時(shí)候誰(shuí)都需要“我這端發(fā)出去一條指令對(duì)面到底處理成什么樣了”這種確定性的返回。這篇文章就把完整做法拆開(kāi)講透從依賴(lài)配置、端點(diǎn)編寫(xiě)、會(huì)話管理到客戶(hù)端封裝、同步回調(diào)協(xié)議、超時(shí)清理和重連保活全部給出可直接抄的代碼和參數(shù)建議。適合正在做即時(shí)通訊、設(shè)備指令下發(fā)或者想把長(zhǎng)連接能力接進(jìn) Spring Boot 業(yè)務(wù)代碼的朋友。1. 項(xiàng)目?jī)r(jià)值拆解為什么“同步回調(diào)”才是真正的業(yè)務(wù)剛需1.1 這個(gè)項(xiàng)目到底解決了什么問(wèn)題普通的 HTTP 接口天然就是“請(qǐng)求-響應(yīng)”模型你調(diào)一次接口拿到返回值事情就結(jié)束了。WebSocket 不一樣它是全雙工長(zhǎng)連接消息推出去之后對(duì)面的響應(yīng)什么時(shí)候回來(lái)、回沒(méi)回來(lái)都是異步事件。這就帶來(lái)一個(gè)很現(xiàn)實(shí)的矛盾Spring Boot 的業(yè)務(wù)代碼大部分是同步風(fēng)格Service 層調(diào)一個(gè)方法、拿到返回值、再往下走大家早就習(xí)慣了這種寫(xiě)法。如果 WebSocket 做成了“發(fā)消息 監(jiān)聽(tīng)回調(diào)”業(yè)務(wù)方就要被拆成兩段一段負(fù)責(zé)發(fā)一段負(fù)責(zé)在回調(diào)里繼續(xù)執(zhí)行代碼邏輯四分五裂排查問(wèn)題也很痛苦。標(biāo)題里的“客戶(hù)端消息同步回調(diào)”解決的就是這個(gè)矛盾對(duì)外暴露的接口仍然像普通方法一樣調(diào)用方傳參、等待、拿結(jié)果內(nèi)部則通過(guò) WebSocket 長(zhǎng)連接異步收發(fā)再靠請(qǐng)求 ID 把響應(yīng)和請(qǐng)求重新對(duì)齊。說(shuō)白了這是把異步通道“偽裝”回同步語(yǔ)義讓長(zhǎng)連接能力和常規(guī)業(yè)務(wù)代碼無(wú)縫銜接。我見(jiàn)過(guò)不少團(tuán)隊(duì)一開(kāi)始做長(zhǎng)連接模塊都是先寫(xiě)廣播 demo結(jié)果一接到真實(shí)業(yè)務(wù)里就發(fā)現(xiàn)各種不對(duì)勁核心原因就是沒(méi)想清楚“請(qǐng)求-響應(yīng)要對(duì)得上號(hào)”這件事。這個(gè)項(xiàng)目本質(zhì)上解決的不是連接問(wèn)題而是“確定性”問(wèn)題。1.2 三個(gè)典型業(yè)務(wù)場(chǎng)景我舉三個(gè)最常見(jiàn)的場(chǎng)景基本覆蓋了標(biāo)題里這套方案會(huì)被用到的位置。第一個(gè)是設(shè)備指令下發(fā)。管理平臺(tái)要控制一臺(tái)終端設(shè)備比如“開(kāi)啟攝像頭”“重啟服務(wù)”“調(diào)整參數(shù)”這種指令光發(fā)出去不算完你一定想知道設(shè)備端是否收到、是否執(zhí)行成功、如果失敗了錯(cuò)誤信息是什么。只有同步回調(diào)才能讓指令下發(fā)服務(wù)保持業(yè)務(wù)一致性不然指令狀態(tài)就只能靠輪詢(xún)數(shù)據(jù)庫(kù)去猜效率低而且實(shí)時(shí)性差。第二個(gè)是在線用戶(hù)操作回執(zhí)。IM 系統(tǒng)里不僅有聊天消息還有“拉黑”“撤回”“已讀回執(zhí)”這類(lèi)帶狀態(tài)的操作。服務(wù)端需要向指定的在線連接推送一條指令并等待用戶(hù)客戶(hù)端應(yīng)答。這種場(chǎng)景如果做成異步廣播你根本不知道被拉黑的那個(gè)用戶(hù)到底有沒(méi)有收到很多業(yè)務(wù)邏輯就沒(méi)法收斂。第三個(gè)是實(shí)時(shí)業(yè)務(wù)場(chǎng)景比如游戲?qū)纸Y(jié)算、交易確認(rèn)、協(xié)同編輯的版本同步。這些業(yè)務(wù)對(duì)結(jié)果返回的實(shí)時(shí)性要求極高HTTP 輪詢(xún)的延遲和開(kāi)銷(xiāo)都不合適WebSocket 長(zhǎng)連接加上同步回調(diào)能把原先“發(fā)了不管”的廣播式通信固化成一個(gè)帶結(jié)果的調(diào)用業(yè)務(wù)代碼寫(xiě)起來(lái)也自然很多?？梢哉f(shuō)只要你的系統(tǒng)里存在“在線設(shè)備或用戶(hù) 需要知道處理結(jié)果”的組合這套方案就是繞不開(kāi)的。1.3 標(biāo)題里的三層架構(gòu)每一層都缺一不可標(biāo)題雖然長(zhǎng)但核心架構(gòu)其實(shí)只有三層。服務(wù)端層負(fù)責(zé)連接管理、會(huì)話存儲(chǔ)、消息接收與推送解決的是“誰(shuí)能連、連上了往哪發(fā)”的問(wèn)題?？蛻?hù)端層是嵌入 Spring Boot 工程里的一個(gè) Java WebSocket 客戶(hù)端組件它對(duì)外暴露業(yè)務(wù)方法對(duì)內(nèi)維護(hù)長(zhǎng)連接狀態(tài)解決的是“業(yè)務(wù)代碼怎么調(diào)用”的問(wèn)題。同步回調(diào)層則是一套消息協(xié)議和等待機(jī)制通過(guò)請(qǐng)求 ID 把響應(yīng)消息和正在等待的請(qǐng)求重新配對(duì)解決的是“結(jié)果怎么回到調(diào)用方手里”的問(wèn)題。三層缺了任何一層都不完整沒(méi)有服務(wù)端客戶(hù)端沒(méi)東西可連沒(méi)有客戶(hù)端業(yè)務(wù)代碼就得裸寫(xiě)一堆 WebSocket 回調(diào)代碼難看還難維護(hù)沒(méi)有同步回調(diào)前面做的所有連接工作都停留在“能通”的水平業(yè)務(wù)根本沒(méi)法用起來(lái)。所以這篇博文會(huì)按這三層逐步拆開(kāi)寫(xiě)順序剛好就是標(biāo)題的閱讀順序。2. 技術(shù)選型與設(shè)計(jì)思路哪些方案能選、我為什么這樣選2.1 服務(wù)端集成 WebSocket 的三種主流姿勢(shì)Spring Boot 里做 WebSocket 服務(wù)端最常見(jiàn)的有三種路子基于ServerEndpoint注解、基于 Spring 的WebSocketHandler注冊(cè)、以及直接集成 Netty。我給它們拉了張對(duì)比表只看它們?cè)凇翱焖偌伞边@個(gè)場(chǎng)景下的表現(xiàn)方案上手難度代碼量生命周期處理適合場(chǎng)景ServerEndpoint注解低最少注解自帶 OnOpen/OnMessage/OnClose/OnError大多數(shù)業(yè)務(wù)項(xiàng)目快速接入沒(méi)問(wèn)題SpringWebSocketHandler中中等實(shí)現(xiàn)接口方法需要和 Spring MVC 深度整合、做攔截握手時(shí)集成 Netty高多自己管理通道和線程模型超大規(guī)模連接、TCP 層定制要求很高時(shí)標(biāo)題里明確寫(xiě)了“快速集成”那我的答案很直接選ServerEndpoint注解方案。原因不復(fù)雜一是它完全由 Java 生態(tài)規(guī)范定義和容器兼容性好代碼量最少二是每個(gè)生命周期回調(diào)都有明確的注解位置不需要去記 Spring 的接口簽名三是不需要額外引入路由邏輯路徑參數(shù)直接寫(xiě)在注解里比如/ws/{userId}在 OnOpen 方法里就能拿到用戶(hù) ID非常貼合業(yè)務(wù)需求。2.2 客戶(hù)端選型與依賴(lài)引入Java 生態(tài)里做 WebSocket 客戶(hù)端我一般直接用 Java-WebSocket 這個(gè)開(kāi)源庫(kù)。它提供了完整的客戶(hù)端實(shí)現(xiàn)連接、發(fā)送、監(jiān)聽(tīng)消息、心跳、重連這些能力都覆蓋了而且它是一個(gè)輕量庫(kù)不需要和 Spring 容器做任何綁定非常適合作為獨(dú)立組件封裝進(jìn)業(yè)務(wù)代碼里。相比之下用 Spring 自帶的WebSocketStompClient也不是不行但它強(qiáng)依賴(lài) STOMP 協(xié)議棧如果你的服務(wù)端是自己實(shí)現(xiàn)的普通 WebSocket 端點(diǎn)沒(méi)必要為握手過(guò)程引入這么重的依賴(lài)。Maven 坐標(biāo)很簡(jiǎn)單服務(wù)端只需要引入 Spring Boot 的 websocket starter客戶(hù)端則額外引入 Java-WebSocket 庫(kù)。注意版本方面選一個(gè)穩(wěn)定版本就行這十年下來(lái)它的 API 變化不大核心用法比較穩(wěn)定。在后面的代碼里我會(huì)把這兩段依賴(lài)分開(kāi)寫(xiě)清楚避免一些人把服務(wù)端和客戶(hù)端的依賴(lài)混在一起出現(xiàn)莫名奇妙的沖突。2.3 “同步回調(diào)”的本質(zhì)把異步消息拉回到同步時(shí)間線要說(shuō)清楚同步回調(diào)我打個(gè)比方。你去銀行柜臺(tái)辦事取了一個(gè)號(hào)碼牌然后坐下來(lái)等廣播叫號(hào)廣播喊到你的號(hào)你再上去拿結(jié)果。這里的號(hào)碼牌就是請(qǐng)求 ID廣播就是 WebSocket 的消息通知等待區(qū)就是你正在阻塞的業(yè)務(wù)線程。實(shí)現(xiàn)上就是用并發(fā)包里的CompletableFuture作為“等待憑證”每次發(fā)送請(qǐng)求前生成一個(gè)全局唯一的請(qǐng)求 ID同時(shí)創(chuàng)建一個(gè)CompletableFuture放進(jìn)一個(gè)ConcurrentHashMap里key 就是請(qǐng)求 ID。消息發(fā)出去之后調(diào)用方執(zhí)行future.get(timeout)阻塞等待。服務(wù)端處理完業(yè)務(wù)后回包時(shí)必須帶上同一個(gè)請(qǐng)求 ID客戶(hù)端在收到消息時(shí)用這個(gè) ID 去 Map 里找到對(duì)應(yīng)的 Future調(diào)用complete()把結(jié)果填進(jìn)去阻塞中的調(diào)用方立刻被喚醒拿到返回值繼續(xù)往下走。這就是“異步轉(zhuǎn)同步”的核心機(jī)制。需要注意這種同步是調(diào)用方線程層面的等待不是整個(gè)系統(tǒng)變成串行服務(wù)端依然是并發(fā)的只是每個(gè)請(qǐng)求在自己的線程里等待自己的結(jié)果彼此之間不干擾。同時(shí)future.get()一定要帶超時(shí)時(shí)間避免服務(wù)端一直不回包把業(yè)務(wù)線程全部打滿(mǎn)這一點(diǎn)在后面的代碼里也會(huì)處理掉。3. 服務(wù)端實(shí)現(xiàn)從握手建連到一對(duì)一消息推送3.1 依賴(lài)與基礎(chǔ)配置服務(wù)端的第一步是加依賴(lài)。Spring Boot 項(xiàng)目里引入spring-boot-starter-websocket就夠了它會(huì)把 WebSocket 相關(guān)的容器能力帶進(jìn)來(lái)同時(shí)和原有的 Web MVC 共存不會(huì)影響普通 HTTP 接口。dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-websocket/artifactId /dependency有了依賴(lài)之后還需要一個(gè)配置類(lèi)注冊(cè)ServerEndpointExporter。這個(gè)類(lèi)非常關(guān)鍵它的作用是掃描所有標(biāo)注了ServerEndpoint的類(lèi)把它們注冊(cè)到底層 WebSocket 容器里讓路徑真正生效。如果少了這一步你寫(xiě)的端點(diǎn)類(lèi)不會(huì)報(bào)錯(cuò)但連接時(shí)永遠(yuǎn)是連接失敗這個(gè)坑不少人都踩過(guò)。Configuration public class WebSocketConfig { Bean public ServerEndpointExporter serverEndpointExporter() { return new ServerEndpointExporter(); } }這一步完成后服務(wù)端的骨架就搭好了。后面寫(xiě)的端點(diǎn)類(lèi)、會(huì)話管理工具、消息中心都會(huì)被 Spring Boot 的自動(dòng)配置能力串起來(lái)。3.2 會(huì)話管理并發(fā)安全是第一個(gè)考點(diǎn)WebSocket 服務(wù)端的核心資源就是Session每個(gè)連接對(duì)應(yīng)一個(gè)會(huì)話我們得把“某個(gè)用戶(hù) ID ? 某個(gè) Session”的對(duì)應(yīng)關(guān)系管理好。因?yàn)?WebSocket 連接的建立和斷開(kāi)是隨時(shí)的可能幾十個(gè)線程同時(shí)在線增刪會(huì)話所以這里必須用并發(fā)安全的容器。我用一個(gè)獨(dú)立的工具類(lèi)來(lái)管理在線會(huì)話內(nèi)部使用ConcurrentHashMap以用戶(hù) ID 為 key以Session對(duì)象為 value。提供四個(gè)最基本的方法加會(huì)話、刪會(huì)話、查會(huì)話、統(tǒng)計(jì)在線數(shù)。為什么單獨(dú)抽一個(gè)類(lèi)因?yàn)槎它c(diǎn)類(lèi)和消息中心都會(huì)用到這份數(shù)據(jù)如果堆在前端類(lèi)里代碼會(huì)越來(lái)越亂還會(huì)出現(xiàn)靜態(tài)變量和實(shí)例變量混淆的問(wèn)題。public class SessionManager { private static final MapString, Session ONLINE_SESSIONS new ConcurrentHashMap(); public static void add(String userId, Session session) { ONLINE_SESSIONS.put(userId, session); } public static void remove(String userId) { ONLINE_SESSIONS.remove(userId); } public static Session get(String userId) { return ONLINE_SESSIONS.get(userId); } public static int onlineCount() { return ONLINE_SESSIONS.size(); } }這里要特別提醒一點(diǎn)同一個(gè)用戶(hù)如果重復(fù)連接比如瀏覽器刷新、客戶(hù)端斷線重連舊的 Session 沒(méi)有正常關(guān)閉Map 里就會(huì)殘留失效連接。所以在add的時(shí)候建議先判斷一下舊 Session 是否還存在如果存在就先關(guān)掉再放新的進(jìn)去。否則你后續(xù)推送時(shí)可能拿到一個(gè)已經(jīng)關(guān)閉的 Session白白增加很多isOpen()判斷。3.3 WebSocket 端點(diǎn)的完整實(shí)現(xiàn)端點(diǎn)類(lèi)直接寫(xiě)在 Spring Boot 工程里加Component和ServerEndpoint兩個(gè)注解。路徑我設(shè)計(jì)成/ws/{userId}這樣連接時(shí)天然帶上了身份信息服務(wù)端不用再通過(guò)消息內(nèi)容去猜是哪個(gè)用戶(hù)。Component ServerEndpoint(/ws/{userId}) public class ChatWebSocketEndpoint { OnOpen public void onOpen(Session session, PathParam(userId) String userId) { SessionManager.add(userId, session); System.out.println(連接建立: userId userId , sessionId session.getId()); } OnMessage public void onMessage(String message, Session session) { System.out.println(收到消息: message); // 這里交給業(yè)務(wù)邏輯處理注意線程池隔離 MessageRouter.route(session, message); } OnClose public void onClose(Session session, PathParam(userId) String userId) { SessionManager.remove(userId); System.out.println(連接關(guān)閉: userId userId); } OnError public void onError(Session session, Throwable error) { error.printStackTrace(); } }這段代碼看起來(lái)簡(jiǎn)單但有幾個(gè)細(xì)節(jié)值得說(shuō)。第一PathParam必須和路徑占位符里的名字一致不然后臺(tái)會(huì)一直打參數(shù)解析異常的日志連接成功率看起來(lái)沒(méi)問(wèn)題但用戶(hù) ID 永遠(yuǎn)是空值。第二onError里最好不要把異常只打日志就算了應(yīng)該把當(dāng)前會(huì)話刪掉避免一個(gè)半死不活的連接還掛在在線列表里占位置。第三onMessage收到的消息體格式建議直接約定為 JSON服務(wù)端不要接收裸字符串否則后面做回調(diào)配對(duì)時(shí)請(qǐng)求 ID 都沒(méi)地方放。3.4 消息中心設(shè)計(jì)對(duì)點(diǎn)推送與廣播有了一對(duì)一的 Session 映射下一步就是封裝一個(gè)消息中心讓業(yè)務(wù)代碼不用直接碰Session對(duì)象。推送消息的核心方法是sendToUser它負(fù)責(zé)根據(jù)用戶(hù) ID 找到 Session檢查連接狀態(tài)然后發(fā)送文本消息。public class MessageCenter { public static void sendToUser(String userId, String message) { Session session SessionManager.get(userId); if (session ! null session.isOpen()) { try { session.getBasicRemote().sendText(message); } catch (IOException e) { SessionManager.remove(userId); e.printStackTrace(); } } } public static void broadcast(String message) { for (String userId : SessionManager.allUserIds()) { sendToUser(userId, message); } } }這里我用了getBasicRemote().sendText()它是同步發(fā)送線程會(huì)等消息寫(xiě)完才返回。如果推送的頻率特別高建議改用getAsyncRemote().sendText()異步發(fā)送不用等 IO 完成但這意味著你要自己處理消息順序和發(fā)送失敗的回調(diào)復(fù)雜度會(huì)上去。另外一個(gè)很多人在推送時(shí)忽略的點(diǎn)同一個(gè)Session不要多個(gè)線程同時(shí)調(diào)用sendText()會(huì)出現(xiàn)消息交錯(cuò)甚至直接拋異常。并發(fā)量大的時(shí)候可以考慮給每個(gè) Session 加一個(gè)寫(xiě)鎖或者用隊(duì)列把所有推送任務(wù)排隊(duì)處理。我在實(shí)際排查過(guò)的問(wèn)題里見(jiàn)過(guò)好幾次“消息偶爾串線”的詭異現(xiàn)象最后定位就是并發(fā)寫(xiě)同一個(gè) Session 造成的。3.5 一個(gè)容易踩的坑Spring 容器與端點(diǎn)實(shí)例的隔離ServerEndpoint的端點(diǎn)類(lèi)和普通的Component有一個(gè)本質(zhì)區(qū)別端點(diǎn)實(shí)例是由 WebSocket 容器創(chuàng)建和管理的每次連接通常對(duì)應(yīng)新的實(shí)例它不是 Spring 的單例。很多人在這里栽過(guò)跟頭在端點(diǎn)類(lèi)里直接Autowired一個(gè) Service結(jié)果發(fā)現(xiàn)是 null因?yàn)?Spring 根本沒(méi)有管理那個(gè)端點(diǎn)實(shí)例。解決辦法是用靜態(tài)字段加 setter 注入的方式把 Spring 容器里的 Service 暴露到端點(diǎn)類(lèi)里。具體做法是端點(diǎn)類(lèi)加Component讓 Spring 掃描到然后在類(lèi)里定義一個(gè)靜態(tài)的 Service 字段和 setter 方法setter 上加Autowired。這樣 Spring 在初始化時(shí)會(huì)把 Service 注入到 setter 里再賦給靜態(tài)字段端點(diǎn)實(shí)例后續(xù)就可以直接使用這個(gè)靜態(tài)字段了。Component ServerEndpoint(/ws/{userId}) public class ChatWebSocketEndpoint { private static MessageService messageService; Autowired public void setMessageService(MessageService service) { ChatWebSocketEndpoint.messageService service; } OnMessage public void onMessage(String message, Session session) { Object result messageService.handle(message); // 繼續(xù)處理 result ... } }這是很多教程里不會(huì)提到的關(guān)鍵細(xì)節(jié)。如果你打電話報(bào)錯(cuò)找不到 Service十有八九就是這里沒(méi)處理好。3.6 服務(wù)端如何配合同步回調(diào)協(xié)議服務(wù)端本身不一定知道“同步回調(diào)”這回事它只需要遵守一條協(xié)議回包時(shí)必須帶上請(qǐng)求里攜帶的reqId。這個(gè)約定是整個(gè)同步機(jī)制成立的基礎(chǔ)。我建議在項(xiàng)目里提前定好統(tǒng)一的消息外殼格式。請(qǐng)求消息和響應(yīng)消息都包含三個(gè)基礎(chǔ)字段reqId表示請(qǐng)求 IDtype表示消息類(lèi)型payload放業(yè)務(wù)數(shù)據(jù)。這樣無(wú)論消息走的是同步請(qǐng)求還是異步事件接收方只要掃一眼type就能分流通過(guò)reqId就能配對(duì)。OnMessage public void onMessage(String message, Session session) { JSONObject request JSON.parseObject(message); String reqId request.getString(reqId); String payload request.getString(payload); Object result messageService.handle(payload); JSONObject response new JSONObject(); response.put(reqId, reqId); response.put(status, 200); response.put(data, result); MessageCenter.sendToUser(getUserId(session), response.toJSONString()); }服務(wù)端回包時(shí)只要找回請(qǐng)求里的reqId填進(jìn)響應(yīng)體客戶(hù)端的同步等待就能被正確喚醒。如果哪天發(fā)現(xiàn)客戶(hù)端一直超時(shí)第一個(gè)排查點(diǎn)就是服務(wù)端有沒(méi)有把reqId原樣返回。這個(gè)看起來(lái)不起眼卻是我見(jiàn)過(guò)最多的失誤原因之一。4. 客戶(hù)端封裝與同步回調(diào)落地4.1 客戶(hù)端核心骨架與連接管理這里的客戶(hù)端指的是嵌入 Spring Boot 工程里的 Java WebSocket 客戶(hù)端組件不是瀏覽器頁(yè)面。它會(huì)去連接遠(yuǎn)端的 WebSocket 服務(wù)然后把一個(gè)“同步請(qǐng)求方法”暴露給上層業(yè)務(wù)調(diào)用。依賴(lài)如下我選用 Java-WebSocket 庫(kù)來(lái)做客戶(hù)端dependency groupIdorg.java-websocket/groupId artifactIdJava-WebSocket/artifactId version1.5.4/version /dependency客戶(hù)端組件核心就是創(chuàng)建一個(gè)WebSocket實(shí)例綁定服務(wù)端地址并在onMessage回調(diào)里集中處理收到的消息。連接建立之后上層業(yè)務(wù)不需要關(guān)心連接對(duì)象只需要調(diào)用我們封裝的同步方法。public class WsClient { private WebSocket socket; public void connect(String serverUrl) throws Exception { socket new WebSocket(new URI(serverUrl)) { Override public void onMessage(String message) { handleResponse(message); } Override public void onClose(int code, String reason, boolean remote) { scheduleReconnect(serverUrl); } }; socket.connect(); // 等待連接完成 CountDownLatch latch new CountDownLatch(1); // 這里可做握手完成的標(biāo)志位簡(jiǎn)化示例直接按需等待 } }有一點(diǎn)要注意connect()是非阻塞的真正建立連接需要幾毫秒到幾十毫秒不等。所以如果你的業(yè)務(wù)方剛調(diào)用完connect()立刻就去發(fā)消息很可能會(huì)拋“連接未打開(kāi)”的異常。一個(gè)穩(wěn)妥的做法是在 onOpen 回調(diào)里放一個(gè)門(mén)閂等連接建立后再開(kāi)放發(fā)消息。我實(shí)際做的時(shí)候是在內(nèi)部維護(hù)了一個(gè)AtomicBoolean connected標(biāo)志每次發(fā)送前檢查沒(méi)連上就阻塞等待一會(huì)兒。4.2 請(qǐng)求-響應(yīng)對(duì)齊的關(guān)鍵結(jié)構(gòu)整個(gè)同步回調(diào)機(jī)制最關(guān)鍵的數(shù)據(jù)結(jié)構(gòu)有兩塊一個(gè)是請(qǐng)求 ID 生成器一個(gè)是等待響應(yīng)的 Future 容器。請(qǐng)求 ID 我用AtomicLong自增序列來(lái)生成配合一個(gè)簡(jiǎn)單的前綴保證在本進(jìn)程內(nèi)唯一。如果多個(gè)實(shí)例部署建議用 UUID 或者在前綴里加上實(shí)例標(biāo)識(shí)避免把不同實(shí)例的請(qǐng)求響應(yīng)匹配錯(cuò)。public class WsClient { private static final AtomicLong REQ_SEQ new AtomicLong(0); private final MapString, CompletableFutureString pendingMap new ConcurrentHashMap(); private String generateReqId() { return REQ_ REQ_SEQ.incrementAndGet(); } }pendingMap就是那個(gè)“等待區(qū)”。每個(gè)請(qǐng)求發(fā)出前往 Map 里放一個(gè)CompletableFuture收到對(duì)應(yīng)響應(yīng)時(shí)從這個(gè) Map 里取出 Future把它完成掉。Map 用ConcurrentHashMap是因?yàn)榭赡苡卸鄠€(gè)業(yè)務(wù)線程同時(shí)發(fā)起請(qǐng)求、多個(gè)消息回調(diào)線程同時(shí)處理響應(yīng)必須在并發(fā)場(chǎng)景下保證安全。4.3 封裝同步請(qǐng)求方法核心方法sendSync的邏輯這樣設(shè)計(jì)生成請(qǐng)求 ID、封裝消息、注冊(cè) Future、發(fā)送消息、阻塞等待、最后無(wú)論如何都要從 Map 里清理掉這個(gè) Future。清理這一步絕對(duì)不能省否則服務(wù)端一直不回包Map 里就會(huì)堆積大量的 Future 對(duì)象時(shí)間一長(zhǎng)就是內(nèi)存泄漏。public String sendSync(String payload, long timeout, TimeUnit unit) throws Exception { String reqId generateReqId(); CompletableFutureString future new CompletableFuture(); pendingMap.put(reqId, future); JSONObject request new JSONObject(); request.put(reqId, reqId); request.put(type, BIZ_REQUEST); request.put(payload, payload); socket.send(request.toJSONString()); try { return future.get(timeout, unit); } finally { pendingMap.remove(reqId); } }這個(gè)方法的調(diào)用方式非常直白業(yè)務(wù)方傳一段 JSON 字符串傳入超時(shí)時(shí)間然后同步拿到服務(wù)端返回的另一個(gè) JSON 字符串。如果服務(wù)端那邊處理時(shí)間大致可預(yù)估建議把超時(shí)設(shè)置成 3 到 5 秒如果包含復(fù)雜的業(yè)務(wù)鏈路可以放寬到 10 秒但超過(guò) 30 秒我一般不建議因?yàn)殚L(zhǎng)時(shí)間阻塞業(yè)務(wù)線程會(huì)讓整個(gè)應(yīng)用的線程池很快耗盡。4.4 心跳保活與斷線重連長(zhǎng)連接有個(gè)天然的敵人網(wǎng)絡(luò)中間設(shè)備和服務(wù)器本身會(huì)清理閑置連接。Tomcat 默認(rèn)會(huì)有空閑連接超時(shí)Nginx 代理也有讀寫(xiě)超時(shí)如果客戶(hù)端長(zhǎng)時(shí)間不發(fā)送任何數(shù)據(jù)連接就會(huì)被對(duì)端靜默關(guān)閉?？蛻?hù)端自己感知不到等到下次發(fā)消息才發(fā)現(xiàn)連接已經(jīng)斷了這時(shí)同步請(qǐng)求就直接卡到超時(shí)。所以客戶(hù)端必須加心跳。我一般用ScheduledExecutorService每隔 10 秒發(fā)送一個(gè)心跳消息消息體統(tǒng)一成{type:PING}。服務(wù)端收到 PING 時(shí)不需要回業(yè)務(wù)數(shù)據(jù)但可以回一個(gè) PONG或者只要服務(wù)端還在正常讀數(shù)據(jù)客戶(hù)端這條心跳就起到了?；钭饔?。private void startHeartbeat() { ScheduledExecutorService executor Executors.newSingleThreadScheduledExecutor(); executor.scheduleAtFixedRate(() - { if (socket ! null socket.isOpen()) { JSONObject ping new JSONObject(); ping.put(type, PING); socket.send(ping.toJSONString()); } }, 10, 10, TimeUnit.SECONDS); }斷線重連方面我會(huì)在onClose里觸發(fā)重連邏輯但重連不能做成無(wú)腦循環(huán)否則服務(wù)端還沒(méi)恢復(fù)客戶(hù)端就會(huì)以極高的頻率瘋狂創(chuàng)建連接把服務(wù)端打得更慘。建議做退避重試第一次 1 秒、第二次 2 秒、第三次 4 秒指數(shù)遞增最多間隔 60 秒服務(wù)端恢復(fù)連接后把重連次數(shù)清零。重連期間所有正在等待的sendSync請(qǐng)求會(huì)因?yàn)槌瑫r(shí)統(tǒng)一失敗上層業(yè)務(wù)要接受這種“短暫不可用”的現(xiàn)實(shí)并在失敗后走自己的重試策略。4.5 回調(diào)協(xié)議的最佳實(shí)踐與參數(shù)建議這部分我總結(jié)了幾個(gè)真正算得上經(jīng)驗(yàn)的東西。第一消息格式必須是 JSON并且在項(xiàng)目一開(kāi)始就定好外殼字段不能今天用reqId明天用requestId這種不一致會(huì)讓整個(gè)調(diào)試過(guò)程變成災(zāi)難。建議統(tǒng)一為reqId、type、payload三個(gè)字段同步響應(yīng)再加一個(gè)status表示結(jié)果狀態(tài)碼。第二sendSync返回的響應(yīng)建議不要直接返回原始字符串最好解析成JSONObject再返回同時(shí)把status非 200 的響應(yīng)當(dāng)成業(yè)務(wù)異常拋出去。這樣上層業(yè)務(wù)代碼拿到的是干凈的結(jié)果而不是還要自己判斷一堆狀態(tài)位。第三請(qǐng)求 ID 最好在客戶(hù)端生成。如果服務(wù)端也可以生成就會(huì)出現(xiàn)兩邊 ID 體系不一致的問(wèn)題調(diào)試時(shí)很容易看花眼?？蛻?hù)端生成后原樣傳給服務(wù)端服務(wù)端原樣回傳鏈路清晰問(wèn)題也好定位。第四同步回調(diào)適合“必須有結(jié)果”的操作但不是所有消息都適合同步。像心跳、狀態(tài)上報(bào)、位置推送這類(lèi)頻率高、不關(guān)心響應(yīng)的消息建議走獨(dú)立的異步通道不要通過(guò)sendSync發(fā)否則線程池重負(fù)載下這些消息也會(huì)排隊(duì)等待白白增加延遲。5. 實(shí)戰(zhàn)避坑與排查實(shí)錄5.1 常見(jiàn)問(wèn)題速查表我把實(shí)際開(kāi)發(fā)中遇到過(guò)的問(wèn)題整理成了一張表按現(xiàn)象排查效率很高現(xiàn)象可能原因解決辦法客戶(hù)端始終連接不上服務(wù)端缺少ServerEndpointExporter注冊(cè)檢查配置類(lèi)是否注冊(cè)了該 Bean連接報(bào) 400 握手失敗路徑參數(shù)不對(duì)或請(qǐng)求路徑與端點(diǎn)不匹配核對(duì)ServerEndpoint路徑和客戶(hù)端連接地址請(qǐng)求 100% 超時(shí)reqId沒(méi)有配對(duì)服務(wù)端響應(yīng)里丟了reqId抓包看服務(wù)端回包字段確認(rèn)reqId原樣返回連接頻繁斷開(kāi)沒(méi)有心跳中間設(shè)備回收空閑連接客戶(hù)端加心跳建議 10 秒一次消息出現(xiàn)錯(cuò)亂或串線同一 Session 被多個(gè)線程并發(fā)寫(xiě)加 Session 寫(xiě)鎖或全部改成異步發(fā)送重連后消息發(fā)不出去重連成功后內(nèi)部連接狀態(tài)沒(méi)更新在 onOpen 回調(diào)里更新連接標(biāo)志位長(zhǎng)時(shí)間運(yùn)行后內(nèi)存增長(zhǎng)明顯pendingMap中超時(shí)的 Future 沒(méi)清理finally塊里執(zhí)行pendingMap.remove(reqId)5.2 幾個(gè)我踩過(guò)的深坑第一個(gè)深坑就是忘記注冊(cè)ServerEndpointExporter。項(xiàng)目啟動(dòng)完全正常沒(méi)有任何報(bào)錯(cuò)但所有 WebSocket 連接都失敗我在瀏覽器控制臺(tái)看到的是 404。排查了半天才發(fā)現(xiàn)是配置類(lèi)里少了一個(gè) Bean這種問(wèn)題報(bào)錯(cuò)信息不直接非常迷惑人。第二個(gè)深坑是瀏覽器里連接 503。如果你前面掛了 Nginx 之類(lèi)的反向代理必須在代理配置里顯式開(kāi)啟協(xié)議升級(jí)并且設(shè)置合理的超時(shí)時(shí)間。很多人本地直連正常一到測(cè)試環(huán)境就失敗基本都是這一層沒(méi)配置。生產(chǎn)環(huán)境還要注意負(fù)載均衡器的會(huì)話保持不然同一個(gè)客戶(hù)端的多次連接會(huì)被轉(zhuǎn)發(fā)到不通的節(jié)點(diǎn)上。第三個(gè)深坑是普通方法里調(diào)用sendSync導(dǎo)致線程池打滿(mǎn)。我當(dāng)時(shí)在業(yè)務(wù)線程里直接調(diào)同步請(qǐng)求結(jié)果服務(wù)端回包變慢幾十個(gè)請(qǐng)求全部卡在future.get()上業(yè)務(wù)線程全部占滿(mǎn)應(yīng)用開(kāi)始出現(xiàn)連環(huán)超時(shí)。后來(lái)給同步請(qǐng)求單獨(dú)隔離了一個(gè)線程池并且嚴(yán)格限制超時(shí)時(shí)間這個(gè)問(wèn)題才算解決。第四個(gè)深坑是關(guān)于 Resend 或者消息重發(fā)。WebSocket 底層不保證消息不丟有時(shí)候服務(wù)端確實(shí)不會(huì)回包客戶(hù)端超時(shí)后會(huì)重試。重試時(shí)不能直接把同一個(gè)reqId重新發(fā)一遍否則如果服務(wù)端前面已經(jīng)處理成功只是回包丟了會(huì)引起業(yè)務(wù)重復(fù)執(zhí)行。建議給每條消息加一個(gè)sequence序號(hào)重試時(shí)用新reqId但在payload里帶上原始業(yè)務(wù)冪等號(hào)服務(wù)端自己決定是否要去重。5.3 分布式多實(shí)例下的拓展思路以上所有方案在單機(jī)模式下都是沒(méi)有問(wèn)題的因?yàn)闀?huì)話信息都存在本地 JVM 的ConcurrentHashMap里。一旦項(xiàng)目從單實(shí)例擴(kuò)展成多實(shí)例部署問(wèn)題就出來(lái)了用戶(hù) A 連接在實(shí)例 1用戶(hù) B 連接在實(shí)例 2實(shí)例 1 想推送消息給 B本地 Session 里根本找不到 B 的連接。常規(guī)解法是把“在線會(huì)話表”從本地內(nèi)存挪到外部存儲(chǔ)比如 Redis 哈希結(jié)構(gòu)里記錄 userId 和節(jié)點(diǎn) ID 的映射。推送時(shí)先查路由表找到目標(biāo)節(jié)點(diǎn)再把消息投遞到對(duì)應(yīng)的實(shí)例由該實(shí)例的本地會(huì)話表完成真正的推送。節(jié)點(diǎn)間通信可以用消息隊(duì)列或者 Redis 發(fā)布訂閱來(lái)做實(shí)例只需消費(fèi)消息并判斷消息里的目標(biāo)節(jié)點(diǎn)是不是自己。同步回調(diào)在這種架構(gòu)下也會(huì)變得復(fù)雜但好在請(qǐng)求 ID 的機(jī)制不受影響只要消息能路由到正確的節(jié)點(diǎn)配對(duì)邏輯依然成立。我通常建議先把單機(jī)版跑通再考慮加分布式路由層不要一上來(lái)就把復(fù)雜度拉滿(mǎn)。最后再分享一點(diǎn)個(gè)人體會(huì)整個(gè)項(xiàng)目跑通之后我最大的感受是同步回調(diào)這個(gè)能力真正難的不是代碼而是協(xié)議先想清楚。你只要把reqId、type、payload、status這套外殼定好后面換服務(wù)端框架、換客戶(hù)端引擎都不傷筋動(dòng)骨。我后來(lái)遇到類(lèi)似的需求第一件事永遠(yuǎn)是先確認(rèn)消息格式而不是先寫(xiě)連接代碼。另外一個(gè)很實(shí)際的心得是同步語(yǔ)義雖好但不要貪只對(duì)業(yè)務(wù)上確實(shí)需要返回結(jié)果的請(qǐng)求用同步像心跳、狀態(tài)廣播這些應(yīng)該走純事件否則你的線程池很快就會(huì)被占滿(mǎn)。要是后續(xù)打算跨集群擴(kuò)展可以把請(qǐng)求 ID 和路由信息塞進(jìn)消息頭再配一個(gè)消息網(wǎng)關(guān)這套客戶(hù)端組件基本不用改服務(wù)端的路由層單獨(dú)替換就行。