我有 7種 實現web實時訊息推播的方案，7種！

技術交流，公眾號：程式設計師小富

大家好，我是小富～

我有一個朋友～

做了一個小破站，現在要實現一個站內信web訊息推播的功能，對，就是下圖這個小紅點，一個很常用的功能。

不過他還沒想好用什麼方式做，這裡我幫他整理了一下幾種方案，並簡單做了實現。

案例下載，記得Star 哦

什麼是訊息推播（push）

推播的場景比較多，比如有人關注我的公眾號，這時我就會收到一條推播訊息，以此來吸引我點選開啟應用。

訊息推播(push)通常是指網站的運營工作等人員，通過某種工具對使用者當前網頁或移動裝置APP進行的主動訊息推播。

訊息推播一般又分為web端訊息推播和行動端訊息推播。

上邊的這種屬於行動端訊息推播，web端訊息推播常見的諸如站內信、未讀郵件數量、監控報警數量等，應用的也非常廣泛。

在具體實現之前，咱們再來分析一下前邊的需求，其實功能很簡單，只要觸發某個事件（主動分享了資源或者後臺主動推播訊息），web頁面的通知小紅點就會實時的+1就可以了。

通常在伺服器端會有若干張訊息推播表，用來記錄使用者觸發不同事件所推播不同型別的訊息，前端主動查詢（拉）或者被動接收（推）使用者所有未讀的訊息數。

訊息推播無非是推（push）和拉（pull）兩種形式，下邊我們逐個瞭解下。

短輪詢

輪詢(polling)應該是實現訊息推播方案中最簡單的一種，這裡我們暫且將輪詢分為短輪詢和長輪詢。

短輪詢很好理解，指定的時間間隔，由瀏覽器向伺服器發出HTTP請求，伺服器實時返回未讀訊息資料給使用者端，瀏覽器再做渲染顯示。

一個簡單的JS定時器就可以搞定，每秒鐘請求一次未讀訊息數介面，返回的資料展示即可。

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

效果還是可以的，短輪詢實現固然簡單，缺點也是顯而易見，由於推播資料並不會頻繁變更，無論後端此時是否有新的訊息產生，使用者端都會進行請求，勢必會對伺服器端造成很大壓力，浪費頻寬和伺服器資源。

長輪詢

長輪詢是對上邊短輪詢的一種改進版本，在儘可能減少對伺服器資源浪費的同時，保證訊息的相對實時性。長輪詢在中介軟體中應用的很廣泛，比如Nacos和apollo設定中心，訊息佇列kafka、RocketMQ中都有用到長輪詢。

Nacos設定中心互動模型是push還是pull？一文中我詳細介紹過Nacos長輪詢的實現原理，感興趣的小夥伴可以瞅瞅。

這次我使用apollo設定中心實現長輪詢的方式，應用了一個類DeferredResult，它是在servelet3.0後經過Spring封裝提供的一種非同步請求機制，直意就是延遲結果。

DeferredResult可以允許容器執行緒快速釋放佔用的資源，不阻塞請求執行緒，以此接受更多的請求提升系統的吞吐量，然後啟動非同步工作執行緒處理真正的業務邏輯，處理完成呼叫DeferredResult.setResult(200)提交響應結果。

下邊我們用長輪詢來實現訊息推播。

因為一個ID可能會被多個長輪詢請求監聽，所以我採用了guava包提供的Multimap結構存放長輪詢，一個key可以對應多個value。一旦監聽到key發生變化，對應的所有長輪詢都會響應。前端得到非請求超時的狀態碼，知曉資料變更，主動查詢未讀訊息數介面，更新頁面資料。

@Controller
@RequestMapping("/polling")
public class PollingController {

    // 存放監聽某個Id的長輪詢集合
    // 執行緒同步結構
    public static Multimap<String, DeferredResult<String>> watchRequests = Multimaps.synchronizedMultimap(HashMultimap.create());

    /**
     * 公眾號：程式設計師小富
     * 設定監聽
     */
    @GetMapping(path = "watch/{id}")
    @ResponseBody
    public DeferredResult<String> watch(@PathVariable String id) {
        // 延遲物件設定超時時間
        DeferredResult<String> deferredResult = new DeferredResult<>(TIME_OUT);
        // 非同步請求完成時移除 key，防止記憶體溢位
        deferredResult.onCompletion(() -> {
            watchRequests.remove(id, deferredResult);
        });
        // 註冊長輪詢請求
        watchRequests.put(id, deferredResult);
        return deferredResult;
    }

    /**
     * 公眾號：程式設計師小富
     * 變更資料
     */
    @GetMapping(path = "publish/{id}")
    @ResponseBody
    public String publish(@PathVariable String id) {
        // 資料變更 取出監聽ID的所有長輪詢請求，並一一響應處理
        if (watchRequests.containsKey(id)) {
            Collection<DeferredResult<String>> deferredResults = watchRequests.get(id);
            for (DeferredResult<String> deferredResult : deferredResults) {
                deferredResult.setResult("我更新了" + new Date());
            }
        }
        return "success";
    }

當請求超過設定的超時時間，會丟擲AsyncRequestTimeoutException異常，這裡直接用@ControllerAdvice全域性捕獲統一返回即可，前端獲取約定好的狀態碼後再次發起長輪詢請求，如此往復呼叫。

@ControllerAdvice
public class AsyncRequestTimeoutHandler {

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

我們來測試一下，首先頁面發起長輪詢請求/polling/watch/10086監聽訊息更變，請求被掛起，不變更資料直至超時，再次發起了長輪詢請求；緊接著手動變更資料/polling/publish/10086，長輪詢得到響應，前端處理業務邏輯完成後再次發起請求，如此迴圈往復。

長輪詢相比於短輪詢在效能上提升了很多，但依然會產生較多的請求，這是它的一點不完美的地方。

iframe流

iframe流就是在頁面中插入一個隱藏的<iframe>標籤，通過在src中請求訊息數量API介面，由此在伺服器端和使用者端之間建立一條長連線，伺服器端持續向iframe傳輸資料。

傳輸的資料通常是HTML、或是內嵌的javascript指令碼，來達到實時更新頁面的效果。

這種方式實現簡單，前端只要一個<iframe>標籤搞定了

<iframe src="/iframe/message" style="display:none"></iframe>

伺服器端直接組裝html、js指令碼資料向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>");
        }
    }
}

但我個人不推薦，因為它在瀏覽器上會顯示請求未載入完，圖示會不停旋轉，簡直是強迫症殺手。

SSE (我的方式)

很多人可能不知道，伺服器端向用戶端推播訊息，其實除了可以用WebSocket這種耳熟能詳的機制外，還有一種伺服器傳送事件(Server-sent events)，簡稱SSE。

SSE它是基於HTTP協定的，我們知道一般意義上的HTTP協定是無法做到伺服器端主動向使用者端推播訊息的，但SSE是個例外，它變換了一種思路。

SSE在伺服器和使用者端之間開啟一個單向通道，伺服器端響應的不再是一次性的封包而是text/event-stream型別的資料流資訊，在有資料變更時從伺服器流式傳輸到使用者端。

整體的實現思路有點類似於線上視訊播放，視訊流會連續不斷的推播到瀏覽器，你也可以理解成，使用者端在完成一次用時很長（網路不暢）的下載。

SSE與WebSocket作用相似，都可以建立伺服器端與瀏覽器之間的通訊，實現伺服器端向用戶端推播訊息，但還是有些許不同：

• SSE 是基於HTTP協定的，它們不需要特殊的協定或伺服器實現即可工作；WebSocket需單獨伺服器來處理協定。
• SSE 單向通訊，只能由伺服器端向用戶端單向通訊；webSocket全雙工通訊，即通訊的雙方可以同時傳送和接受資訊。
• SSE 實現簡單開發成本低，無需引入其他元件；WebSocket傳輸資料需做二次解析，開發門檻高一些。
• SSE 預設支援斷線重連；WebSocket則需要自己實現。
• SSE 只能傳送文字訊息，二進位制資料需要經過編碼後傳送；WebSocket預設支援傳送二進位制資料。

SSE 與 WebSocket 該如何選擇？

技術並沒有好壞之分，只有哪個更合適

SSE好像一直不被大家所熟知，一部分原因是出現了WebSockets，這個提供了更豐富的協定來執行雙向、全雙工通訊。對於遊戲、即時通訊以及需要雙向近乎實時更新的場景，擁有雙向通道更具吸引力。

但是，在某些情況下，不需要從使用者端傳送資料。而你只需要一些伺服器操作的更新。比如：站內信、未讀訊息數、狀態更新、股票行情、監控數量等場景，SEE不管是從實現的難易和成本上都更加有優勢。此外，SSE 具有WebSockets在設計上缺乏的多種功能，例如：自動重新連線、事件ID和傳送任意事件的能力。

前端只需進行一次HTTP請求，帶上唯一ID，開啟事件流，監聽伺服器端推播的事件就可以了

<script>
    let source = null;
    let userId = 7777
    if (window.EventSource) {
        // 建立連線
        source = new EventSource('http://localhost:7777/sse/sub/'+userId);
        setMessageInnerHTML("連線使用者=" + userId);
        /**
         * 連線一旦建立，就會觸發open事件
         * 另一種寫法：source.onopen = function (event) {}
         */
        source.addEventListener('open', function (e) {
            setMessageInnerHTML("建立連線。。。");
        }, false);
        /**
         * 使用者端收到伺服器發來的資料
         * 另一種寫法：source.onmessage = function (event) {}
         */
        source.addEventListener('message', function (e) {
            setMessageInnerHTML(e.data);
        });
    } else {
        setMessageInnerHTML("你的瀏覽器不支援SSE");
    }
</script>

伺服器端的實現更簡單，建立一個SseEmitter物件放入sseEmitterMap進行管理

private static Map<String, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();

/**
 * 建立連線
 *
 * @date: 2022/7/12 14:51
 * @auther: 公眾號：程式設計師小富
 */
public static SseEmitter connect(String userId) {
    try {
        // 設定超時時間，0表示不過期。預設30秒
        SseEmitter sseEmitter = new SseEmitter(0L);
        // 註冊回撥
        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("建立新的sse連線異常，當前使用者：{}", userId);
    }
    return null;
}

/**
 * 給指定使用者傳送訊息
 *
 * @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);
        }
    }
}

我們模擬伺服器端推播訊息，看下使用者端收到了訊息，和我們預期的效果一致。

注意： SSE不支援IE瀏覽器，對其他主流瀏覽器相容性做的還不錯。

MQTT

什麼是 MQTT協定？

MQTT 全稱(Message Queue Telemetry Transport)：一種基於釋出/訂閱（publish/subscribe）模式的輕量級通訊協定，通過訂閱相應的主題來獲取訊息，是物聯網（Internet of Thing）中的一個標準傳輸協定。

該協定將訊息的釋出者（publisher）與訂閱者（subscriber）進行分離，因此可以在不可靠的網路環境中，為遠端連線的裝置提供可靠的訊息服務，使用方式與傳統的MQ有點類似。

TCP協定位於傳輸層，MQTT 協定位於應用層，MQTT 協定構建於TCP/IP協定上，也就是說只要支援TCP/IP協定棧的地方，都可以使用MQTT協定。

為什麼要用 MQTT協定？

MQTT協定為什麼在物聯網（IOT）中如此受偏愛？而不是其它協定，比如我們更為熟悉的 HTTP協定呢？

• 首先HTTP協定它是一種同步協定，使用者端請求後需要等待伺服器的響應。而在物聯網（IOT）環境中，裝置會很受制於環境的影響，比如頻寬低、網路延遲高、網路通訊不穩定等，顯然非同步訊息協定更為適合IOT應用程式。

• HTTP是單向的，如果要獲取訊息使用者端必須發起連線，而在物聯網（IOT）應用程式中，裝置或感測器往往都是使用者端，這意味著它們無法被動地接收來自網路的命令。

• 通常需要將一條命令或者訊息，傳送到網路上的所有裝置上。HTTP要實現這樣的功能不但很困難，而且成本極高。

具體的MQTT協定介紹和實踐，這裡我就不再贅述了，大家可以參考我之前的兩篇文章，裡邊寫的也都很詳細了。

MQTT協定的介紹

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

MQTT實現訊息推播

未讀訊息（小紅點），前端 與 RabbitMQ 實時訊息推播實踐，賊簡單~

Websocket

websocket應該是大家都比較熟悉的一種實現訊息推播的方式，上邊我們在講SSE的時候也和websocket進行過比較。

WebSocket是一種在TCP連線上進行全雙工通訊的協定，建立使用者端和伺服器之間的通訊渠道。瀏覽器和伺服器僅需一次握手，兩者之間就直接可以建立永續性的連線，並進行雙向資料傳輸。

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

<!-- 引入websocket -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

伺服器端使用@ServerEndpoint註解標註當前類為一個websocket伺服器，使用者端可以通過ws://localhost:7777/webSocket/10086來連線到WebSocket伺服器端。

@Component
@Slf4j
@ServerEndpoint("/websocket/{userId}")
public class WebSocketServer {
    //與某個使用者端的連線對談，需要通過它來給使用者端傳送資料
    private Session session;
    private static final CopyOnWriteArraySet<WebSocketServer> webSockets = new CopyOnWriteArraySet<>();
    // 用來存線上連線數
    private static final Map<String, Session> sessionPool = new HashMap<String, Session>();
    /**
     * 公眾號：程式設計師小富
     * 連結成功呼叫的方法
     */
    @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訊息: 有新的連線，總數為:" + webSockets.size());
        } catch (Exception e) {
        }
    }
    /**
     * 公眾號：程式設計師小富
     * 收到使用者端訊息後呼叫的方法
     */
    @OnMessage
    public void onMessage(String message) {
        log.info("websocket訊息: 收到使用者端訊息:" + message);
    }
    /**
     * 公眾號：程式設計師小富
     * 此為單點訊息
     */
    public void sendOneMessage(String userId, String message) {
        Session session = sessionPool.get(userId);
        if (session != null && session.isOpen()) {
            try {
                log.info("websocket消: 單點訊息:" + message);
                session.getAsyncRemote().sendText(message);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

前端初始化開啟WebSocket連線，並監聽連線狀態，接收伺服器端資料或向伺服器端傳送資料。

<script>
    var ws = new WebSocket('ws://localhost:7777/webSocket/10086');
    // 獲取連線狀態
    console.log('ws連線狀態：' + ws.readyState);
    //監聽是否連線成功
    ws.onopen = function () {
        console.log('ws連線狀態：' + ws.readyState);
        //連線成功則傳送一個資料
        ws.send('test1');
    }
    // 接聽伺服器發回的資訊並處理展示
    ws.onmessage = function (data) {
        console.log('接收到來自伺服器的訊息：');
        console.log(data);
        //完成通訊後關閉WebSocket連線
        ws.close();
    }
    // 監聽連線關閉事件
    ws.onclose = function () {
        // 監聽整個過程中websocket的狀態
        console.log('ws連線狀態：' + ws.readyState);
    }
    // 監聽並處理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連線，之後就可以進行雙向通訊了，效果還不錯

自定義推播

上邊我們給我出了6種方案的原理和程式碼實現，但在實際業務開發過程中，不能盲目的直接拿過來用，還是要結合自身系統業務的特點和實際場景來選擇合適的方案。

推播最直接的方式就是使用第三推播平臺，畢竟錢能解決的需求都不是問題，無需複雜的開發運維，直接可以使用，省時、省力、省心，像goEasy、極光推播都是很不錯的三方服務商。

一般大型公司都有自研的訊息推播平臺，像我們本次實現的web站內信只是平臺上的一個觸點而已，簡訊、郵件、微信公眾號、小程式凡是可以觸達到使用者的渠道都可以接入進來。

訊息推播系統內部是相當複雜的，諸如訊息內容的維護稽核、圈定推播人群、觸達過濾攔截（推播的規則頻次、時段、數量、黑白名單、關鍵詞等等）、推播失敗補償非常多的模組，技術上涉及到巨量資料量、高並行的場景也很多。所以我們今天的實現方式在這個龐大的系統面前只是小打小鬧。

Github地址

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

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