1 問題背景

擴寬自己的知識廣度，研究一下web實時消息推送

2 前言

1. 文章參考自Web 實時消息推送的 7 種實現(xiàn)方案
2. 針對一些比較重要的方式，我都會盡量敲出一份完整的demo代碼，享受其中的編程樂趣。
3. 在SSE方式中，筆者延申思考，將他應(yīng)用于電商支付的場景中，給出了比較合理的解決方案，但并未在生產(chǎn)環(huán)境中驗證，仍待考證。

3 什么是消息推送

消息推送是指服務(wù)端將消息推送給客戶端。常見的場景有：有人關(guān)注公眾號，公眾號推送消息給關(guān)注者；站內(nèi)消息通知；未讀郵件數(shù)量；監(jiān)控告警數(shù)量等等。

4 短輪詢

常見的http請求即是短輪詢，由客戶端發(fā)起請求，服務(wù)端接收請求并同步實時處理，最后返回數(shù)據(jù)給客戶端。

5 長輪詢

短輪詢的異步方式即是長輪詢，異步在哪里？客戶端發(fā)起請求，web容器（比如tomcat）安排子線程去處理這些請求，將這些請求交給服務(wù)端后，無需阻塞等待結(jié)果，tomcat會立即安排該子線程理其他請求 ，tomcat以此接收更多的請求提升系統(tǒng)的吞吐量。服務(wù)端處理完請求再返回數(shù)據(jù)給客戶端。

5.1 demo代碼

因為一個ID可能會被多個長輪詢請求監(jiān)聽，所以采用了guava包提供的Multimap結(jié)構(gòu)存放長輪詢，一個key可以對應(yīng)多個value。一旦監(jiān)聽到key發(fā)生變化，對應(yīng)的所有長輪詢都會響應(yīng)。

引入guava依賴

<dependency><groupId>com.google.guava</groupId><artifactId>guava</artifactId><version>31.1-jre</version>
</dependency>

處理請求的接口：

package com.ganzalang.gmall.sse.controller;import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimap;
import com.google.common.collect.Multimaps;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.context.request.async.DeferredResult;import java.time.LocalDateTime;
import java.util.Collection;
import java.util.Date;@Controller
@RequestMapping("/polling")
public class PollingController {/*** 關(guān)于 DeferredResult 還有一個很重要的點：請求的處理線程（即 tomcat 線程池的線程）不會等到 DeferredResult#setResult() 被調(diào)用才釋放，而是直接釋放了。* 而 DeferredResult 的做法就類似僅把事情安排好，不會管事情做好沒，tomcat 線程就釋放走了，注意此時不會給請求方（如瀏覽器）任何響應(yīng)，而是將請求存放在一邊，* 咱先不管它，等后面有結(jié)果了再把之前的請求拿來，把值響應(yīng)給請求方。*/// 存放監(jiān)聽某個Id的長輪詢集合// 線程同步結(jié)構(gòu)public static Multimap<String, DeferredResult<String>> watchRequests = Multimaps.synchronizedMultimap(HashMultimap.create());public static final long TIME_OUT = 100000;/*** 設(shè)置監(jiān)聽*/@GetMapping(path = "watch/{id}")@ResponseBodypublic DeferredResult<String> watch(@PathVariable String id) {// 延遲對象設(shè)置超時時間DeferredResult<String> deferredResult = new DeferredResult<>(TIME_OUT);// 異步請求完成時移除 key，防止內(nèi)存溢出deferredResult.onCompletion(() -> {watchRequests.remove(id, deferredResult);});// 注冊長輪詢請求watchRequests.put(id, deferredResult);return deferredResult;}/*** 變更數(shù)據(jù)*/@GetMapping(path = "publish/{id}")@ResponseBodypublic String publish(@PathVariable String id) {// 數(shù)據(jù)變更 取出監(jiān)聽ID的所有長輪詢請求，并一一響應(yīng)處理if (watchRequests.containsKey(id)) {Collection<DeferredResult<String>> deferredResults = watchRequests.get(id);for (DeferredResult<String> deferredResult : deferredResults) {deferredResult.setResult("我更新了" + LocalDateTime.now());}}return "success";}/*** 監(jiān)聽器的數(shù)量*/@GetMapping(path = "listener/num")@ResponseBodypublic int num() {return watchRequests.size();}
}

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

package com.ganzalang.gmall.sse.exception.handler;import org.springframework.http.HttpStatus;
import org.springframework.web.bind.annotation.ControllerAdvice;
import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.context.request.async.AsyncRequestTimeoutException;@ControllerAdvice
public class AsyncRequestTimeoutHandler {@ResponseStatus(HttpStatus.NOT_MODIFIED)@ResponseBody@ExceptionHandler(AsyncRequestTimeoutException.class)public String asyncRequestTimeoutHandler(AsyncRequestTimeoutException e) {System.out.println("異步請求超時");return "304";}
}

測試：

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

6 iframe流

在頁面中插入一個隱藏的<iframe>標(biāo)簽，通過在src中請求消息數(shù)量API接口，由此在服務(wù)端和客戶端之間創(chuàng)建一條長連接，服務(wù)端持續(xù)向iframe傳輸數(shù)據(jù)。傳輸?shù)臄?shù)據(jù)通常是html，js腳本。

6.1 demo代碼

筆者打算使用一個頁面來展示效果，因此需要引入一個freemarker依賴

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-freemarker</artifactId>
</dependency>

yml中配置freemarker：

spring:freemarker:suffix: .ftlcontent-type: text/htmlcharset: UTF-8cache: false# ftl頁面存放的路徑template-loader-path: classpath:/templates/

寫一個ftl頁面：

ftl頁面源碼如下：

<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><title>IFRAME</title>
</head>
<body>
<iframe src="/iframe/message" style="display:none"></iframe>
<div><h1>clock</h1><span id="clock"></span><h1>count</h1><span id="count"></span>
</div>
</body>
</html>

服務(wù)端的代碼：

package com.ganzalang.gmall.sse.controller;import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.time.LocalDateTime;
import java.util.concurrent.atomic.AtomicInteger;@Controller
@RequestMapping("/iframe")
@Slf4j
public class IframeController {/*** 訪問首頁** @param request* @return* @throws IOException*/@GetMapping("/index")public String index(HttpServletRequest request) throws IOException {log.info("iframe-index");return "iframe-index";}/*** 返回消息** @param response* @throws IOException* @throws InterruptedException*/@GetMapping(path = "message")public void message(HttpServletResponse response) throws IOException, InterruptedException {AtomicInteger count = new AtomicInteger(1);while (true) {log.info("current time:{}", LocalDateTime.now());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.getAndIncrement() + "\";" + "</script>");}}
}

測試：

訪問http://localhost:8033/iframe/index即可，大家會發(fā)現(xiàn)這樣非常占用服務(wù)器資源，服務(wù)端會很卡。并且客戶端還會一直在loading，如下圖所示：

7 SSE

Server-sent events，簡稱SSE。SSE在服務(wù)器和客戶端之間打開一個單向通道，服務(wù)端響應(yīng)的不再是一次性的數(shù)據(jù)包而是text/event-stream類型的數(shù)據(jù)流信息，在有數(shù)據(jù)變更時從服務(wù)器流式傳輸?shù)娇蛻舳?。SSE有如下幾個特點：

1. 基于HTTP
2. 單向通信
3. 實現(xiàn)簡單，無需引入其組件
4. 默認(rèn)支持?jǐn)嗑€重連 （服務(wù)端重啟，客戶端會重新發(fā)送連接請求，這是天生解決服務(wù)端發(fā)版的問題啊）
5. 只能傳送文本信息

7.1 demo代碼

SSE同樣是使用頁面展示效果，需要添加一些freemarker相關(guān)東西，具體細(xì)節(jié)可見第6.1節(jié)

頁面代碼：

頁面源碼：

<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><title>IFRAME</title>
</head>
<body>
<div id="message"></div>
<script>let source = null;// 獲取url中userId參數(shù)的值。如url=http://localhost:8033/sse/index?userId=1111var userId = window.location.search.substring(1).split('&')[0].split('=')[1];// 判斷當(dāng)前客戶端（瀏覽器）是否支持SSE，有些瀏覽器不是默認(rèn)支持SSE的if (window.EventSource) {// 建立連接source = new EventSource('http://localhost:8033/sse/sub/'+userId);document.getElementById("message").innerHTML += "連接用戶=" + userId + "<br>";/*** 連接一旦建立，就會觸發(fā)open事件* 另一種寫法：source.onopen = function (event) {}*/source.addEventListener('open', function (e) {document.getElementById("message").innerHTML += "建立連接。。。<br>";}, false);/*** 客戶端收到服務(wù)器發(fā)來的數(shù)據(jù)* 另一種寫法：source.onmessage = function (event) {}*/source.addEventListener('message', function (e) {document.getElementById("message").innerHTML += e.data + "<br>";});} else {document.getElementById("message").innerHTML += "你的瀏覽器不支持SSE<br>";}
</script></body>
</html>

服務(wù)端代碼：

package com.ganzalang.gmall.sse.controller;import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Controller;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;@Controller
@RequestMapping("/sse")
@Slf4j
public class SseController {private static Map<String, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();/*** 推送消息給客戶端** @param userId* @param msg* @throws IOException* @throws InterruptedException*/@GetMapping(path = "/message/{userId}")public void message(@PathVariable("userId") String userId, @RequestParam(value = "msg", required = false)String msg) throws IOException, InterruptedException {String message = StringUtils.isEmpty(msg) ? "pay success" : msg;sendMessage(userId, message);}/*** 查詢當(dāng)前的sse連接數(shù)量** @return* @throws IOException* @throws InterruptedException*/@GetMapping(path = "/num")@ResponseBodypublic String num() throws IOException, InterruptedException {return String.valueOf(sseEmitterMap.keySet().size());}@GetMapping(path = "/del/{userId}")@ResponseBodypublic String num(@PathVariable("userId") String userId) throws IOException, InterruptedException {sseEmitterMap.remove(userId);return "success";}/*** 開啟sse連接** @param userId* @return* @throws IOException* @throws InterruptedException*/@GetMapping("/sub/{userId}")@ResponseBodypublic SseEmitter sub(@PathVariable("userId") String userId) throws IOException, InterruptedException {SseEmitter sseEmitter = connect(userId);log.info("userId={}, result:{}", userId, "Pay success");return sseEmitter;}/*** 訪問sse首頁** @return* @throws IOException*/@GetMapping("/index")public String index() throws IOException {log.info("sse-index");return "sse-index";}/*** 創(chuàng)建連接** @date: 2022/7/12 14:51*/public static SseEmitter connect(String userId) {try {// 設(shè)置超時時間，0表示不過期。默認(rèn)30秒SseEmitter sseEmitter = new SseEmitter(0L);// 注冊回調(diào)sseEmitter.onCompletion(() -> removeUser(userId));sseEmitter.onError((e) -> log.error("exception:{}", e.getMessage(), e));sseEmitter.onTimeout(() -> removeUser(userId));sseEmitterMap.put(userId, sseEmitter);return sseEmitter;} catch (Exception e) {log.info("創(chuàng)建新的sse連接異常，當(dāng)前用戶：{}", userId);}return null;}/*** 給指定用戶發(fā)送消息** @date: 2022/7/12 14:51*/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);}}}/*** 移除對應(yīng)的客戶端連接* * @param userId*/private static void removeUser(String userId) {sseEmitterMap.remove(userId);}
}

測試：

訪問http://localhost:8033/sse/index?userId=1111注冊客戶端連接，此處記為index頁面。

瀏覽器另開一個tab頁，訪問http://localhost:8033/sse/message/1111?msg=haha，然后去index頁面看，會發(fā)現(xiàn)有消息展示出來了：

當(dāng)服務(wù)端重啟，客戶端會自動重連，即index頁面的那個http請求會再次發(fā)給服務(wù)端

7.2 生產(chǎn)環(huán)境的應(yīng)用 （重要）

筆者做支付比較多，該sse方式也可以用于做支付結(jié)果的消息通知（一般都是用短輪詢做查詢，查詢支付結(jié)果；那么現(xiàn)在可以使用SSE方式）。針對應(yīng)用于生產(chǎn)環(huán)境，筆者認(rèn)為有如下幾點需要注意：

1. 由前面服務(wù)端代碼可見，服務(wù)端需要在內(nèi)存中保存客戶端的連接（那個sseEmitterMap）。在服務(wù)端是集群的情況下，接收客戶端請求的服務(wù)端節(jié)點的內(nèi)存中，并不一定就有客戶端的連接，此處可以使用Redis的發(fā)布訂閱功能，通知存有客戶端連接的服務(wù)端節(jié)點進(jìn)行發(fā)消息。除了Redis發(fā)布訂閱，還能通過Redis+RPC做一個精準(zhǔn)調(diào)用，Redis可以存儲Map<客戶端連接的唯一標(biāo)識, 服務(wù)端節(jié)點IP>，拿到IP后通過RPC進(jìn)行精準(zhǔn)調(diào)用，詳情可以見服務(wù)端實時推送技術(shù)之SSE（Server-Send Events）。

2. 服務(wù)端節(jié)點中的每個客戶端連接都需要做超時處理，超時則將連接從內(nèi)存中移除，否則會發(fā)生OOM。

3. 假如服務(wù)端發(fā)版，內(nèi)存中的所有客戶端連接都會丟失，但無需擔(dān)憂，因為客戶端默認(rèn)會重連。

8 MQTT

MQTT方式需要借助消息隊列來實現(xiàn)，其實相當(dāng)于常規(guī)的生產(chǎn)者消費者模式。因?qū)崿F(xiàn)起來比較復(fù)雜，（需要搭建MQ），筆者此處暫不研究MQTT具體實現(xiàn)。