1 問(wèn)題背景

擴(kuò)寬自己的知識(shí)廣度，研究一下web實(shí)時(shí)消息推送

2 前言

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

3 什么是消息推送

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

4 短輪詢

常見(jiàn)的http請(qǐng)求即是短輪詢，由客戶端發(fā)起請(qǐng)求，服務(wù)端接收請(qǐng)求并同步實(shí)時(shí)處理，最后返回?cái)?shù)據(jù)給客戶端。

5 長(zhǎng)輪詢

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

5.1 demo代碼

因?yàn)橐粋€(gè)ID可能會(huì)被多個(gè)長(zhǎng)輪詢請(qǐng)求監(jiān)聽(tīng)，所以采用了guava包提供的Multimap結(jié)構(gòu)存放長(zhǎng)輪詢，一個(gè)key可以對(duì)應(yīng)多個(gè)value。一旦監(jiān)聽(tīng)到key發(fā)生變化，對(duì)應(yīng)的所有長(zhǎng)輪詢都會(huì)響應(yīng)。

引入guava依賴

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

處理請(qǐng)求的接口：

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 還有一個(gè)很重要的點(diǎn)：請(qǐng)求的處理線程（即 tomcat 線程池的線程）不會(huì)等到 DeferredResult#setResult() 被調(diào)用才釋放，而是直接釋放了。* 而 DeferredResult 的做法就類(lèi)似僅把事情安排好，不會(huì)管事情做好沒(méi)，tomcat 線程就釋放走了，注意此時(shí)不會(huì)給請(qǐng)求方（如瀏覽器）任何響應(yīng)，而是將請(qǐng)求存放在一邊，* 咱先不管它，等后面有結(jié)果了再把之前的請(qǐng)求拿來(lái)，把值響應(yīng)給請(qǐng)求方。*/// 存放監(jiān)聽(tīng)某個(gè)Id的長(zhǎng)輪詢集合// 線程同步結(jié)構(gòu)public static Multimap<String, DeferredResult<String>> watchRequests = Multimaps.synchronizedMultimap(HashMultimap.create());public static final long TIME_OUT = 100000;/*** 設(shè)置監(jiān)聽(tīng)*/@GetMapping(path = "watch/{id}")@ResponseBodypublic DeferredResult<String> watch(@PathVariable String id) {// 延遲對(duì)象設(shè)置超時(shí)時(shí)間DeferredResult<String> deferredResult = new DeferredResult<>(TIME_OUT);// 異步請(qǐng)求完成時(shí)移除 key，防止內(nèi)存溢出deferredResult.onCompletion(() -> {watchRequests.remove(id, deferredResult);});// 注冊(cè)長(zhǎng)輪詢請(qǐng)求watchRequests.put(id, deferredResult);return deferredResult;}/*** 變更數(shù)據(jù)*/@GetMapping(path = "publish/{id}")@ResponseBodypublic String publish(@PathVariable String id) {// 數(shù)據(jù)變更 取出監(jiān)聽(tīng)I(yíng)D的所有長(zhǎng)輪詢請(qǐng)求，并一一響應(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)聽(tīng)器的數(shù)量*/@GetMapping(path = "listener/num")@ResponseBodypublic int num() {return watchRequests.size();}
}

當(dāng)請(qǐng)求超過(guò)設(shè)置的超時(shí)時(shí)間，會(huì)拋出AsyncRequestTimeoutException異常，這里直接用@ControllerAdvice全局捕獲統(tǒng)一返回即可，前端獲取約定好的狀態(tài)碼后再次發(fā)起長(zhǎng)輪詢請(qǐng)求，如此往復(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("異步請(qǐng)求超時(shí)");return "304";}
}

測(cè)試：

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

6 iframe流

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

6.1 demo代碼

筆者打算使用一個(gè)頁(yè)面來(lái)展示效果，因此需要引入一個(gè)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頁(yè)面存放的路徑template-loader-path: classpath:/templates/

寫(xiě)一個(gè)ftl頁(yè)面：

ftl頁(yè)面源碼如下：

<!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 {/*** 訪問(wèn)首頁(yè)** @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>");}}
}

測(cè)試：

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

7 SSE

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

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

7.1 demo代碼

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

頁(yè)面代碼：

頁(yè)面源碼：

<!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>";/*** 連接一旦建立，就會(huì)觸發(fā)open事件* 另一種寫(xiě)法：source.onopen = function (event) {}*/source.addEventListener('open', function (e) {document.getElementById("message").innerHTML += "建立連接。。。<br>";}, false);/*** 客戶端收到服務(wù)器發(fā)來(lái)的數(shù)據(jù)* 另一種寫(xiě)法：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";}/*** 開(kāi)啟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;}/*** 訪問(wèn)sse首頁(yè)** @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è)置超時(shí)時(shí)間，0表示不過(guò)期。默認(rèn)30秒SseEmitter sseEmitter = new SseEmitter(0L);// 注冊(cè)回調(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);}}}/*** 移除對(duì)應(yīng)的客戶端連接* * @param userId*/private static void removeUser(String userId) {sseEmitterMap.remove(userId);}
}

測(cè)試：

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

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

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

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

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

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

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

3. 假如服務(wù)端發(fā)版，內(nèi)存中的所有客戶端連接都會(huì)丟失，但無(wú)需擔(dān)憂，因?yàn)榭蛻舳四J(rèn)會(huì)重連。

8 MQTT

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