我有一個(gè)朋友～

做了一個(gè)小破站，現(xiàn)在要實(shí)現(xiàn)一個(gè)站內(nèi)信web消息推送的功能，對，就是下圖這個(gè)小紅點(diǎn)，一個(gè)很常用的功能。

【資料圖】

不過他還沒想好用什么方式做，這里我?guī)退砹艘幌聨追N方案，并簡單做了實(shí)現(xiàn)。

什么是消息推送（push）

推送的場景比較多，比如有人關(guān)注我的公眾號，這時(shí)我就會收到一條推送消息，以此來吸引我點(diǎn)擊打開應(yīng)用。

消息推送(push)通常是指網(wǎng)站的運(yùn)營工作等人員，通過某種工具對用戶當(dāng)前網(wǎng)頁或移動設(shè)備APP進(jìn)行的主動消息推送。

消息推送一般又分為web端消息推送?和移動端消息推送。

上邊的這種屬于移動端消息推送，web端消息推送常見的諸如站內(nèi)信、未讀郵件數(shù)量、監(jiān)控報(bào)警數(shù)量等，應(yīng)用的也非常廣泛。

在具體實(shí)現(xiàn)之前，咱們再來分析一下前邊的需求，其實(shí)功能很簡單，只要觸發(fā)某個(gè)事件（主動分享了資源或者后臺主動推送消息），web頁面的通知小紅點(diǎn)就會實(shí)時(shí)的+1就可以了。

通常在服務(wù)端會有若干張消息推送表，用來記錄用戶觸發(fā)不同事件所推送不同類型的消息，前端主動查詢（拉）或者被動接收（推）用戶所有未讀的消息數(shù)。

消息推送無非是推（push?）和拉（pull）兩種形式，下邊我們逐個(gè)了解下。

短輪詢

輪詢(polling?)應(yīng)該是實(shí)現(xiàn)消息推送方案中最簡單的一種，這里我們暫且將輪詢分為短輪詢?和長輪詢。

短輪詢很好理解，指定的時(shí)間間隔，由瀏覽器向服務(wù)器發(fā)出HTTP請求，服務(wù)器實(shí)時(shí)返回未讀消息數(shù)據(jù)給客戶端，瀏覽器再做渲染顯示。

一個(gè)簡單的JS定時(shí)器就可以搞定，每秒鐘請求一次未讀消息數(shù)接口，返回的數(shù)據(jù)展示即可。

setInterval(() => { // 方法請求 messageCount().then((res) => { if (res.code === 200) { this.messageCount = res.data } })}, 1000);

效果還是可以的，短輪詢實(shí)現(xiàn)固然簡單，缺點(diǎn)也是顯而易見，由于推送數(shù)據(jù)并不會頻繁變更，無論后端此時(shí)是否有新的消息產(chǎn)生，客戶端都會進(jìn)行請求，勢必會對服務(wù)端造成很大壓力，浪費(fèi)帶寬和服務(wù)器資源。

長輪詢

長輪詢是對上邊短輪詢的一種改進(jìn)版本，在盡可能減少對服務(wù)器資源浪費(fèi)的同時(shí)，保證消息的相對實(shí)時(shí)性。長輪詢在中間件中應(yīng)用的很廣泛，比如Nacos?和apollo?配置中心，消息隊(duì)列kafka?、RocketMQ中都有用到長輪詢。

Nacos配置中心交互模型是push還是pull？?一文中我詳細(xì)介紹過Nacos長輪詢的實(shí)現(xiàn)原理，感興趣的小伙伴可以瞅瞅。

這次我使用apollo?配置中心實(shí)現(xiàn)長輪詢的方式，應(yīng)用了一個(gè)類DeferredResult?，它是在servelet3.0后經(jīng)過Spring封裝提供的一種異步請求機(jī)制，直意就是延遲結(jié)果。

DeferredResult?可以允許容器線程快速釋放占用的資源，不阻塞請求線程，以此接受更多的請求提升系統(tǒng)的吞吐量，然后啟動異步工作線程處理真正的業(yè)務(wù)邏輯，處理完成調(diào)用DeferredResult.setResult(200)提交響應(yīng)結(jié)果。

下邊我們用長輪詢來實(shí)現(xiàn)消息推送。

因?yàn)橐粋€(gè)ID可能會被多個(gè)長輪詢請求監(jiān)聽，所以我采用了guava?包提供的Multimap結(jié)構(gòu)存放長輪詢，一個(gè)key可以對應(yīng)多個(gè)value。一旦監(jiān)聽到key發(fā)生變化，對應(yīng)的所有長輪詢都會響應(yīng)。前端得到非請求超時(shí)的狀態(tài)碼，知曉數(shù)據(jù)變更，主動查詢未讀消息數(shù)接口，更新頁面數(shù)據(jù)。

@Controller@RequestMapping("/polling")public class PollingController { // 存放監(jiān)聽某個(gè)Id的長輪詢集合 // 線程同步結(jié)構(gòu) public static Multimap> watchRequests = Multimaps.synchronizedMultimap(HashMultimap.create()); /** * 公眾號：程序員小富 * 設(shè)置監(jiān)聽 */ @GetMapping(path = "watch/{id}") @ResponseBody public DeferredResult watch(@PathVariable String id) { // 延遲對象設(shè)置超時(shí)時(shí)間 DeferredResult deferredResult = new DeferredResult<>(TIME_OUT); // 異步請求完成時(shí)移除 key，防止內(nèi)存溢出 deferredResult.onCompletion(() -> { watchRequests.remove(id, deferredResult); }); // 注冊長輪詢請求 watchRequests.put(id, deferredResult); return deferredResult; } /** * 公眾號：程序員小富 * 變更數(shù)據(jù) */ @GetMapping(path = "publish/{id}") @ResponseBody public String publish(@PathVariable String id) { // 數(shù)據(jù)變更 取出監(jiān)聽ID的所有長輪詢請求，并一一響應(yīng)處理 if (watchRequests.containsKey(id)) { Collection> deferredResults = watchRequests.get(id); for (DeferredResult deferredResult : deferredResults) { deferredResult.setResult("我更新了" + new Date()); } } return "success"; }

當(dāng)請求超過設(shè)置的超時(shí)時(shí)間，會拋出AsyncRequestTimeoutException?異常，這里直接用@ControllerAdvice全局捕獲統(tǒng)一返回即可，前端獲取約定好的狀態(tài)碼后再次發(fā)起長輪詢請求，如此往復(fù)調(diào)用。

@ControllerAdvicepublic class AsyncRequestTimeoutHandler { @ResponseStatus(HttpStatus.NOT_MODIFIED) @ResponseBody @ExceptionHandler(AsyncRequestTimeoutException.class) public String asyncRequestTimeoutHandler(AsyncRequestTimeoutException e) { System.out.println("異步請求超時(shí)"); return "304"; }}

我們來測試一下，首先頁面發(fā)起長輪詢請求/polling/watch/10086?監(jiān)聽消息更變，請求被掛起，不變更數(shù)據(jù)直至超時(shí)，再次發(fā)起了長輪詢請求；緊接著手動變更數(shù)據(jù)/polling/publish/10086，長輪詢得到響應(yīng)，前端處理業(yè)務(wù)邏輯完成后再次發(fā)起請求，如此循環(huán)往復(fù)。

長輪詢相比于短輪詢在性能上提升了很多，但依然會產(chǎn)生較多的請求，這是它的一點(diǎn)不完美的地方。

iframe流

iframe流就是在頁面中插入一個(gè)隱藏的

服務(wù)端直接組裝html、js腳本數(shù)據(jù)向response寫入就行了

@Controller@RequestMapping("/iframe")public class IframeController { @GetMapping(path = "message") public void message(HttpServletResponse response) throws IOException, InterruptedException { while (true) { response.setHeader("Pragma", "no-cache"); response.setDateHeader("Expires", 0); response.setHeader("Cache-Control", "no-cache,no-store"); response.setStatus(HttpServletResponse.SC_OK); response.getWriter().print(" <script type=\"text/javascript\">\n" + "parent.document.getElementById("clock").innerHTML = \"" + count.get() + "\";" + "parent.document.getElementById("count").innerHTML = \"" + count.get() + "\";" + "</script>"); } }}

但我個(gè)人不推薦，因?yàn)樗跒g覽器上會顯示請求未加載完，圖標(biāo)會不停旋轉(zhuǎn)，簡直是強(qiáng)迫癥殺手。

SSE (我的方式)

很多人可能不知道，服務(wù)端向客戶端推送消息，其實(shí)除了可以用WebSocket?這種耳熟能詳?shù)臋C(jī)制外，還有一種服務(wù)器發(fā)送事件(Server-sent events?)，簡稱SSE。

SSE?它是基于HTTP協(xié)議的，我們知道一般意義上的HTTP協(xié)議是無法做到服務(wù)端主動向客戶端推送消息的，但SSE是個(gè)例外，它變換了一種思路。

SSE在服務(wù)器和客戶端之間打開一個(gè)單向通道，服務(wù)端響應(yīng)的不再是一次性的數(shù)據(jù)包而是text/event-stream類型的數(shù)據(jù)流信息，在有數(shù)據(jù)變更時(shí)從服務(wù)器流式傳輸?shù)娇蛻舳恕?/p>

整體的實(shí)現(xiàn)思路有點(diǎn)類似于在線視頻播放，視頻流會連續(xù)不斷的推送到瀏覽器，你也可以理解成，客戶端在完成一次用時(shí)很長（網(wǎng)絡(luò)不暢）的下載。

SSE?與WebSocket作用相似，都可以建立服務(wù)端與瀏覽器之間的通信，實(shí)現(xiàn)服務(wù)端向客戶端推送消息，但還是有些許不同：

SSE 是基于HTTP協(xié)議的，它們不需要特殊的協(xié)議或服務(wù)器實(shí)現(xiàn)即可工作；WebSocket需單獨(dú)服務(wù)器來處理協(xié)議。SSE 單向通信，只能由服務(wù)端向客戶端單向通信；webSocket全雙工通信，即通信的雙方可以同時(shí)發(fā)送和接受信息。SSE 實(shí)現(xiàn)簡單開發(fā)成本低，無需引入其他組件；WebSocket傳輸數(shù)據(jù)需做二次解析，開發(fā)門檻高一些。SSE 默認(rèn)支持?jǐn)嗑€重連；WebSocket則需要自己實(shí)現(xiàn)。SSE 只能傳送文本消息，二進(jìn)制數(shù)據(jù)需要經(jīng)過編碼后傳送；WebSocket默認(rèn)支持傳送二進(jìn)制數(shù)據(jù)。SSE 與 WebSocket 該如何選擇？

技術(shù)并沒有好壞之分，只有哪個(gè)更合適

SSE好像一直不被大家所熟知，一部分原因是出現(xiàn)了WebSockets，這個(gè)提供了更豐富的協(xié)議來執(zhí)行雙向、全雙工通信。對于游戲、即時(shí)通信以及需要雙向近乎實(shí)時(shí)更新的場景，擁有雙向通道更具吸引力。

但是，在某些情況下，不需要從客戶端發(fā)送數(shù)據(jù)。而你只需要一些服務(wù)器操作的更新。比如：站內(nèi)信、未讀消息數(shù)、狀態(tài)更新、股票行情、監(jiān)控?cái)?shù)量等場景，SEE?不管是從實(shí)現(xiàn)的難易和成本上都更加有優(yōu)勢。此外，SSE 具有WebSockets?在設(shè)計(jì)上缺乏的多種功能，例如：自動重新連接?、事件ID?和發(fā)送任意事件的能力。

前端只需進(jìn)行一次HTTP請求，帶上唯一ID，打開事件流，監(jiān)聽服務(wù)端推送的事件就可以了

<script> let source = null; let userId = 7777 if (window.EventSource) { // 建立連接 source = new EventSource("http://localhost:7777/sse/sub/"+userId); setMessageInnerHTML("連接用戶=" + userId); /** * 連接一旦建立，就會觸發(fā)open事件 * 另一種寫法：source.onopen = function (event) {} */ source.addEventListener("open", function (e) { setMessageInnerHTML("建立連接。。。"); }, false); /** * 客戶端收到服務(wù)器發(fā)來的數(shù)據(jù) * 另一種寫法：source.onmessage = function (event) {} */ source.addEventListener("message", function (e) { setMessageInnerHTML(e.data); }); } else { setMessageInnerHTML("你的瀏覽器不支持SSE"); }</script>

服務(wù)端的實(shí)現(xiàn)更簡單，創(chuàng)建一個(gè)SseEmitter?對象放入sseEmitterMap進(jìn)行管理

private static Map sseEmitterMap = new ConcurrentHashMap<>();/** * 創(chuàng)建連接 * * @date: 2022/7/12 14:51 * @auther: 公眾號：程序員小富 */public static SseEmitter connect(String userId) { try { // 設(shè)置超時(shí)時(shí)間，0表示不過期。默認(rèn)30秒 SseEmitter sseEmitter = new SseEmitter(0L); // 注冊回調(diào) sseEmitter.onCompletion(completionCallBack(userId)); sseEmitter.onError(errorCallBack(userId)); sseEmitter.onTimeout(timeoutCallBack(userId)); sseEmitterMap.put(userId, sseEmitter); count.getAndIncrement(); return sseEmitter; } catch (Exception e) { log.info("創(chuàng)建新的sse連接異常，當(dāng)前用戶：{}", userId); } return null;}/** * 給指定用戶發(fā)送消息 * * @date: 2022/7/12 14:51 * @auther: 公眾號：程序員小富 */public static void sendMessage(String userId, String message) { if (sseEmitterMap.containsKey(userId)) { try { sseEmitterMap.get(userId).send(message); } catch (IOException e) { log.error("用戶[{}]推送異常:{}", userId, e.getMessage()); removeUser(userId); } }}

我們模擬服務(wù)端推送消息，看下客戶端收到了消息，和我們預(yù)期的效果一致。

注意：SSE不支持IE?瀏覽器，對其他主流瀏覽器兼容性做的還不錯(cuò)。

MQTT

什么是 MQTT協(xié)議？

MQTT?全稱(Message Queue Telemetry Transport)：一種基于發(fā)布/訂閱（publish?/subscribe?）模式的輕量級?通訊協(xié)議，通過訂閱相應(yīng)的主題來獲取消息，是物聯(lián)網(wǎng)（Internet of Thing）中的一個(gè)標(biāo)準(zhǔn)傳輸協(xié)議。

該協(xié)議將消息的發(fā)布者（publisher?）與訂閱者（subscriber）進(jìn)行分離，因此可以在不可靠的網(wǎng)絡(luò)環(huán)境中，為遠(yuǎn)程連接的設(shè)備提供可靠的消息服務(wù)，使用方式與傳統(tǒng)的MQ有點(diǎn)類似。

TCP?協(xié)議位于傳輸層，MQTT?協(xié)議位于應(yīng)用層，MQTT?協(xié)議構(gòu)建于TCP/IP?協(xié)議上，也就是說只要支持TCP/IP?協(xié)議棧的地方，都可以使用MQTT協(xié)議。

為什么要用 MQTT協(xié)議？

MQTT?協(xié)議為什么在物聯(lián)網(wǎng)（IOT）中如此受偏愛？而不是其它協(xié)議，比如我們更為熟悉的HTTP協(xié)議呢？

首先HTTP?協(xié)議它是一種同步協(xié)議，客戶端請求后需要等待服務(wù)器的響應(yīng)。而在物聯(lián)網(wǎng)（IOT）環(huán)境中，設(shè)備會很受制于環(huán)境的影響，比如帶寬低、網(wǎng)絡(luò)延遲高、網(wǎng)絡(luò)通信不穩(wěn)定等，顯然異步消息協(xié)議更為適合IOT應(yīng)用程序。HTTP是單向的，如果要獲取消息客戶端必須發(fā)起連接，而在物聯(lián)網(wǎng)（IOT）應(yīng)用程序中，設(shè)備或傳感器往往都是客戶端，這意味著它們無法被動地接收來自網(wǎng)絡(luò)的命令。通常需要將一條命令或者消息，發(fā)送到網(wǎng)絡(luò)上的所有設(shè)備上。HTTP要實(shí)現(xiàn)這樣的功能不但很困難，而且成本極高。

具體的MQTT協(xié)議介紹和實(shí)踐，這里我就不再贅述了，大家可以參考我之前的兩篇文章，里邊寫的也都很詳細(xì)了。

MQTT協(xié)議的介紹

我也沒想到 springboot + rabbitmq 做智能家居，會這么簡單

MQTT實(shí)現(xiàn)消息推送

未讀消息（小紅點(diǎn)），前端 與 RabbitMQ 實(shí)時(shí)消息推送實(shí)踐，賊簡單~

Websocket

websocket應(yīng)該是大家都比較熟悉的一種實(shí)現(xiàn)消息推送的方式，上邊我們在講SSE的時(shí)候也和websocket進(jìn)行過比較。

WebSocket是一種在TCP連接上進(jìn)行全雙工通信的協(xié)議，建立客戶端和服務(wù)器之間的通信渠道。瀏覽器和服務(wù)器僅需一次握手，兩者之間就直接可以創(chuàng)建持久性的連接，并進(jìn)行雙向數(shù)據(jù)傳輸。

圖片源于網(wǎng)絡(luò)

springboot整合websocket，先引入websocket相關(guān)的工具包，和SSE相比額外的開發(fā)成本。

org.springframework.boot spring-boot-starter-websocket

服務(wù)端使用@ServerEndpoint?注解標(biāo)注當(dāng)前類為一個(gè)websocket服務(wù)器，客戶端可以通過ws://localhost:7777/webSocket/10086來連接到WebSocket服務(wù)器端。

@Component@Slf4j@ServerEndpoint("/websocket/{userId}")public class WebSocketServer { //與某個(gè)客戶端的連接會話，需要通過它來給客戶端發(fā)送數(shù)據(jù) private Session session; private static final CopyOnWriteArraySet webSockets = new CopyOnWriteArraySet<>(); // 用來存在線連接數(shù) private static final Map sessionPool = new HashMap(); /** * 公眾號：程序員小富 * 鏈接成功調(diào)用的方法 */ @OnOpen public void onOpen(Session session, @PathParam(value = "userId") String userId) { try { this.session = session; webSockets.add(this); sessionPool.put(userId, session); log.info("websocket消息: 有新的連接，總數(shù)為:" + webSockets.size()); } catch (Exception e) { } } /** * 公眾號：程序員小富 * 收到客戶端消息后調(diào)用的方法 */ @OnMessage public void onMessage(String message) { log.info("websocket消息: 收到客戶端消息:" + message); } /** * 公眾號：程序員小富 * 此為單點(diǎn)消息 */ public void sendOneMessage(String userId, String message) { Session session = sessionPool.get(userId); if (session != null && session.isOpen()) { try { log.info("websocket消: 單點(diǎn)消息:" + message); session.getAsyncRemote().sendText(message); } catch (Exception e) { e.printStackTrace(); } } }}

前端初始化打開WebSocket連接，并監(jiān)聽連接狀態(tài)，接收服務(wù)端數(shù)據(jù)或向服務(wù)端發(fā)送數(shù)據(jù)。

<script> var ws = new WebSocket("ws://localhost:7777/webSocket/10086"); // 獲取連接狀態(tài) console.log("ws連接狀態(tài)：" + ws.readyState); //監(jiān)聽是否連接成功 ws.onopen = function () { console.log("ws連接狀態(tài)：" + ws.readyState); //連接成功則發(fā)送一個(gè)數(shù)據(jù) ws.send("test1"); } // 接聽服務(wù)器發(fā)回的信息并處理展示 ws.onmessage = function (data) { console.log("接收到來自服務(wù)器的消息："); console.log(data); //完成通信后關(guān)閉WebSocket連接 ws.close(); } // 監(jiān)聽連接關(guān)閉事件 ws.onclose = function () { // 監(jiān)聽整個(gè)過程中websocket的狀態(tài) console.log("ws連接狀態(tài)：" + ws.readyState); } // 監(jiān)聽并處理error事件 ws.onerror = function (error) { console.log(error); } function sendMessage() { var content = $("#message").val(); $.ajax({ url: "/socket/publish?userId=10086&message=" + content, type: "GET", data: { "id": "7777", "content": content }, success: function (data) { console.log(data) } }) }</script>

頁面初始化建立websocket連接，之后就可以進(jìn)行雙向通信了，效果還不錯(cuò)

自定義推送

上邊我們給我出了6種方案的原理和代碼實(shí)現(xiàn)，但在實(shí)際業(yè)務(wù)開發(fā)過程中，不能盲目的直接拿過來用，還是要結(jié)合自身系統(tǒng)業(yè)務(wù)的特點(diǎn)和實(shí)際場景來選擇合適的方案。

推送最直接的方式就是使用第三推送平臺，畢竟錢能解決的需求都不是問題，無需復(fù)雜的開發(fā)運(yùn)維，直接可以使用，省時(shí)、省力、省心，像goEasy、極光推送都是很不錯(cuò)的三方服務(wù)商。

一般大型公司都有自研的消息推送平臺，像我們本次實(shí)現(xiàn)的web站內(nèi)信只是平臺上的一個(gè)觸點(diǎn)而已，短信、郵件、微信公眾號、小程序凡是可以觸達(dá)到用戶的渠道都可以接入進(jìn)來。

圖片來源于網(wǎng)絡(luò)

消息推送系統(tǒng)內(nèi)部是相當(dāng)復(fù)雜的，諸如消息內(nèi)容的維護(hù)審核、圈定推送人群、觸達(dá)過濾攔截（推送的規(guī)則頻次、時(shí)段、數(shù)量、黑白名單、關(guān)鍵詞等等）、推送失敗補(bǔ)償非常多的模塊，技術(shù)上涉及到大數(shù)據(jù)量、高并發(fā)的場景也很多。所以我們今天的實(shí)現(xiàn)方式在這個(gè)龐大的系統(tǒng)面前只是小打小鬧。

Github地址

文中所提到的案例我都一一的做了實(shí)現(xiàn)，整理放在了Github上，覺得有用就Star一下吧！

傳送門：https://github.com/chengxy-nds/Springboot-Notebook/tree/master/springboot-realtime-data

標(biāo)簽：