基于發布訂閱機制的實時消息系統設計與實現

收稿日期：2023-06-30

DOI：10.19850/j.cnki.2096-4706.2023.21.007

摘" 要：基于Apache Kafka消息代理框架使用成本高和基于私有協議導致的兼容性不好等問題，提出了以WebSocket作為通信協議的發布訂閱機制實時消息系統的設計方案。以聊天系統為例，使用Socket.IO實現后端服務，使用Redis的發布訂閱機制來構建實時消息系統，同時提出了全局ID、消息令牌等模塊的設計思路。首先進行了方案技術選型，具體包括消息傳遞模型與協議選型、后端框架選型，然后闡述了系統架構設計和主要模塊實現，最后進行了系統測試。測試結果表明系統運行穩定，實時連接性能良好。

關鍵詞：發布訂閱；實時消息系統；Web Socket；Socket.IO

中圖分類號：TP312" " 文獻標識碼：A" " 文章編號：2096-4706（2023）21-0028-05

Design and Implementation of Real-time Messaging System Based on

Publish-Subscribe Mechanism

XING Ruyi

（Xuzhou Finance and Economics Branch of Jiangsu Union Technical Institute， Xuzhou" 221008， China）

Abstract： Based on the high cost of Apache Kafka message proxy framework and the poor compatibility caused by private protocol， a design scheme of publish-subscribe mechanism real-time message system with WebSocket as communication protocol is proposed. Taking the chat system as an example， using Socket. IO to implement backend services， using Redis' publish-subscribe mechanism to build a real-time messaging system， and proposing design ideas for modules such as global ID and message token. Firstly， the technical selection of the scheme is carried out， including the selection of messaging model and protocol， and the selection of back-end framework. Then， the system architecture design and the realization of main modules are expounded， and finally the system is tested. The test results show that the system runs stably and the real-time connection performance is good.

Keywords： publish-subscribe; real-time messaging system; Web Socket; Socket.IO

0" 引" 言

Apache Kafka是當前使用廣泛和優秀的分布式消息平臺，其提供了基于發布訂閱機制的消息系統，可以實現消息生產者生產消息發送到消息系統，消息系統投遞給消費者；提供了大容量存儲系統，支持大數據量存儲和歷史數據存儲；提供了實時數據管道，支持建立面向MySQL等數據源的實時數據管道，繼而為其他系統提供數據服務；提供了流計算能力，支持與Spark、Flink等分布式計算引擎結合，構建實時流應用，實時處理大規模數據。

張積存等[1]提出在車輛大數據分析系統中將Kafka作為系統消息總線，搭建Kafka集群實現跨系統的海量數據實時傳輸。蘇玉釗等[2]針對負載均衡導致消費者消息處理下降問題，提出基于區塊鏈的Kafka共識算法優化機制。孫海勇等[3]提出基于Kafka消息隊列與MongoDB數據庫構建實時數據采集平臺，實現加工數據實時接收與本地存儲。陳樂等[4]提出了將Nginx與Kafka直連實現日志數據實時采集，后端作為消費者獲取日志消息進行實時計算。

雖然上Kafka平臺在功能設計和架構設計上非常優秀，但對于構建中小型實時消息系統時還存在一些不足，具體包括：平臺學習難度高，Kafka平臺內部進行了大量復雜的設計，在使用前需要明確平臺內部的設計架構與概念特征。平臺采用私有協議，其自行設計了一套二進制通信協議，與現有開放的互聯網協議不兼容，需要使用相應組件才能與開放協議結合起來。基于以上考慮，本文以構建實時聊天系統為例，提出以WebSocket作為通信協議、使用Socket.IO實現后端服務、使用Redis實現發布訂閱機制來構建實時消息系統，同時提出了全局ID、消息令牌等模塊的設計思路。

1" 技術選型

1.1" 消息傳遞模式選型

互聯網、移動互聯網和物聯網的迅速發展促進了實時消息系統的應用，日常使用的即時聊天軟件（微信等）、在線聊天應用、在線客服系統等都是實時消息系統。通常來看，對于面向用戶使用的系統應用的“實時”是指響應時間在250毫秒以內、操作系統中“硬件實時”要求延遲小于5毫秒、實時游戲中則要求低于50毫秒，因此實時系統也可定義為低延遲系統。實時消息系統在設計時要求要具備高可靠性、可擴展性、有序性等特點，在設計上需考慮協議選擇、技術選型、緩存設計等內容。事件驅動是消息系統的主流的編程模式，而事件驅動的典型應用即發布訂閱模式。

消息傳遞模式選型上，發布訂閱模式是分布式系統在進行架構設計時常用的一種設計模式，可用于在不同組件或服務之間進行異步通信，更有利于異構服務間的通信、服務之間的解耦，更加靈活和可擴展。Amozarrain等[5]提出了發布訂閱機制在移動應用場景下，消息同步方法與實現思路。在發布訂閱模式的體系結構中，發布者是創建和發送消息的組件，訂閱者是接收和使用消息的組件，兩者通過中間的消息代理組件進行通信，消息代理組件維護了多個消息通道，每個通道被多個訂閱者訂閱，也可以接收多個發布者的消息。發布訂閱模式作為一種消息傳遞模式，組件之間具有協議無關、語言無關的特點，適用于發送事件通知、實現分布式緩存、分布式日志記錄、廣播更新等應用場景，適用于構建實時和低延時的消息系統。

1.2" 消息傳遞協議選型

消息傳遞協議選型上，支持低延遲的協議包括MQTT協議、RTMP協議、WebSocket協議等。RTMP協議是一種基于TCP協議、用于在互聯網上實現音頻、視頻數據傳輸的協議；MQTT協議是一種用于物聯網應用中，實現客戶端與服務端低延遲數據傳輸的輕量級協議。HTTP協議構建消息系統，采用HTTP長輪詢方式，采用固定頻率向服務端發送請求，通過提高輪詢頻率減少消息的延遲。但HTTP輪詢會重新連接服務器，在處理高并發請求時會給Web服務器帶來額外壓力和增加帶寬成本。上述三種協議都具有各自應用場景，都不適合應用于實時數據通信場景中。

WebSocket協議可以實現與服務器保持單一、持久連接，從而減輕服務器壓力并解決延遲和帶寬問題。WebSocket協議是建立在TCP協議之上的一種可靠的、雙向和全雙工的通信協議，協議提供了數據有序和錯誤檢查，可以實現客戶端和服務端穩定、低延遲和實時通信。李鵬程等[6]提出了基于WebSocket的MQTT協議，并應用在物聯網環境下構建車輛信息系統。高志輝等[7]提出基于WebSocket構建實時車間數據監測系統的設計與實現?，F有各類文獻表明，WebSocket協議適用于實時系統的開發。

1.3" 服務端實現框架選型

Socket.IO是一個基于事件驅動、用于構建實時Web應用程序Javascript開發庫，是一個方便和強大的實時通信解決方案。其底層支持WebSocket協議、支持客戶端與服務端的雙向通信。Socket.IO具有跨平臺的特性，可運行在多種平臺或設備上；具備良好可靠性，支持網絡中斷自動重連、內置心跳檢測機制；支持多種傳輸方式，包括WebSocket、輪詢與長輪詢，以實現更好的兼容性；具有良好可擴展性，可處理大規模并發連接，支持命名空間概念，可實現連接分組等。

Redis是開源的、基于鍵值對的內存數據存儲器，常被用作數據庫、緩存和消息代理、流引擎等。Redis具有豐富的數據結構，如字符串、散列、列表、集合、位圖、地理坐標等；內置數據自動過期、事務處理、數據持久化等功能；默認將數據存儲在內存中，提供高效數據存儲性能；內置發布訂閱機制的實現，提出了頻道概念，提供了SUBSCRIBE、PUBLISH等命令實現頻道訂閱、信息發布等功能，支持即發即棄、僅扇出等消息模式。胡喜明等[8]提出了將Redis作為數據緩存、Kafka作為消息中間件，使用MQTT協議構建響應式服務推送框架。

2" 系統設計

2.1" 架構設計

基于聊天應用的特征，系統將服務分為三類，分別是無狀態服務、有狀態服務和第三方集成服務。其中無狀態服務是傳統的面向公眾用戶的請求響應，實現用戶登錄、注冊、用戶信息維護等。無狀態服務經Nginx實現負載均衡后，將請求路由相應的服務中。有狀態服務面向于聊天類請求處理，每個客戶端與服務器保持長連接，在使用聊天服務過程中，客戶端與服務器保持對應關系，不會動態切換服務器。

第三方集成服務的主要作用是實現聊天信息的推送，實現將應用的消息推送至郵件、公眾號，等第三方平臺。實現這類功能，平臺需要與第三方平臺進行SDK集成，完成身份信息的綁定、授權等工作。將上述三個服務綜合，結合Redis內存存儲和MySQL關系型持久存儲，整個系統的架構如圖1所示。

2.2" 詳細設計

2.2.1" 全局ID設計

全局唯一ID是分布式系統設計中重點考慮的一個環節。由于消息會經過多個服務之間傳遞，為了鑒別消息的唯一性通常采用分布式全局ID設計方案。Twitter公司開源的Snowflake生成算法是當前流行的全局ID設計方案，其以劃分命名空間的方式將64位字符序列分為多個部分。Snowflake生成算法可實現在1毫秒1臺服務器上產生4 096個有序且不重復的ID，可滿足全局唯一ID的生成需求。

為壓縮全局ID的長度且保持可讀性，文中參照Snowflake算法并進行了修改，將全局ID設計為34位長度的字符。圖2展示了全局ID字符的結構，分別包括時間（14位）、用戶編號（10位）、序列號（10位）。其中時間部分格式為年月日時分秒、序列號部分使用Redis生成。基于此方法生成全局唯一ID使得消息鑒別更具業務屬性，便于數據調試。

2.2.2" 系統組件調用流程設計

實時消息系統在組件上包括服務網關組件、業務服務組件、消息服務組件和數據存儲組件。服務網關負責接收來自終端用戶的請求；業務服務組件處理用戶請求并連接數據存儲組件完成數據管理操作；消息服務組件作為無狀態服務，負責消息請求轉發與處理；數據存儲則包括Redis緩存存儲和MySQL持久存儲。

系統各組件調用流程包括：用戶發起登錄或注冊請求，請求發送至服務網關。服務網關將請求發送給業務服務上，業務服務調用數據存儲完成數據存儲。用戶完成登錄后，系統向消息服務組件發起WebSocket連接，消息服務完成連接，準備接收消息轉發請求。用戶初次連接至消息服務組件后，組件檢查讀取離線消息并發送給客戶端，然后將客戶端連接至實時消息服務組件；系統實時保存聊天消息至Redis，Redis內部維護消息隊列，實現發布訂閱機制。系統采取定時方式將Redis消息保存至MySQL數據庫。最后用戶退出系統前，客戶端主動關閉與消息服務組件的連接。圖3展示了上述業務流程面向于各組件的調用流程。

2.2.3" 訪問令牌設計

訪問令牌解決了用戶登錄后信息認證與鑒權的問題。在消息系統中，系統采用微服務架構設計，為鑒別正常請求，防止惡意請求，通過在請求中加入請求令牌以提高系統安全性。在消息系統中，訪問令牌由認證服務頒發，其他服務將獲取到的令牌發送至認證服務中進行檢驗[9]。認證服務需要處理高并發和大量的訪問令牌生成、令牌驗證工作，底層使用Redis進行保存，以提高工作性能。

消息系統中的令牌是一串經過Base64編碼后的文本字符串，其編碼的原文是對字符串“時間戳+用戶編號+設備編號+會話密鑰”進行AES加密后的文本。其中用戶編號是用戶在系統中的唯一標識，設備編號是用戶登錄的移動端或PC端的設備編號，會話密鑰則是系統預設的加密字符。圖4展示了系統中訪問令牌生成的原理。

2.2.4" 輸入敏感詞過濾

輸入敏感詞過濾是互聯網平臺中對于用戶提交數據的合法、合規性進行檢驗的必備環節，進行敏感詞過濾可以保護用戶隱私，防止用戶敏感信息泄漏或被濫用，防止垃圾信息、惡意廣告等內容傳播，提升用戶體驗和安全性。

文中使用DFA（Deterministic Finite Automaton）算法識別和匹配文本中敏感詞。DFA算法是一種基于確定性轉換的算法，有高效地檢測和過濾敏感詞，具有效率高、內存消耗低、高擴展和匹配精確的特點[10]。文中使用開源的Javascript庫badwords構建，badwords庫內部使用DFA算法實現敏感詞快速匹配和過濾功能。具體步驟為：1）構建敏感詞庫，從網易官網下載敏感詞庫文件；2）使用npm命令安裝badwords模塊；3）構建過濾器Filter，并調用其clean方法實現文本檢測和替換。

3" 系統實現與測試

3.1" 消息廣播功能實現

使用Socket.IO與Redis實現消息廣播功能，使用NodeJS中的http模塊編寫Socket.IO服務端，創建/nofity和/chat兩個命名空間，并啟動服務，關鍵代碼為：

var server = require（'http'）.createServer（）;

var io = require（'Socket.IO'）（server）;

io.of（'/notify）.on（'connection'， socket =gt; {

socket.on（'disconnect'， （） =gt;{}）;

）;

io.of（'/chat）.on（'connection'， socket =gt;{}）;

server.listen（3001， （） =gt;{}）;

測試1：服務連接測試。編寫代碼連接至Redis服務，監聽Redis的廣播消息，根據頻道的不同將消息分發給客戶端，關鍵代碼如下：

redisCli.on（'message'， function（channel， message） {

switch （channle）{

case NOTIFICATION_CHANNEL：

io.of（'/notify）.emit（'message'， message）;

break;

case CHATTING_CHANNEL：

io.of（'/chat）.emit（'message'， message）;

break;

}

}）;

測試2：客戶端測試。使用Socket.IO的JS庫連接至Socket.IO服務，從服務器中相應的命名空間獲取實時消息，關鍵代碼如下：

var socket = io（'/notify）;

socket.on（'message'， function （msg） {

console.log（msg）;

}）;

由于WebSocket是有狀態的，客戶端與服務端建立長連接后，在服務端需要維護和管理用戶會話。服務端內部維護了一個哈希表，哈希表中將“用戶編號、用戶會話對象”作為鍵值對保存，其中用戶會話對象內部維護了其使用的頻道列表信息。為了解決用戶創建長連接數量過多問題，系統設置當頻道超過一定數量后，早期建立的頻道將自動關閉。

3.2" 測試結果分析

測試環境使用兩臺工作端分別作為客戶端和服務端。工作站物理配置為4核CPU、16 GB內存，千兆網卡局域網連接。使用apache ab命令模擬20個并發用戶，每個用戶建立連接5萬個，模擬并發連接量為100萬個，使用ss命令查看服務器的連接數和內存使用情，如圖5所示。實驗證明，采用WebSocket協議和Redis構建發布訂閱消息系統在性能上能夠處理大規模在線請求。

4" 結" 論

移動互聯網的普及推動了實時消息系統的開發與應用。實時消息系統可以幫助企業實現消息實時觸達、消息跨終端同步等業務場景。文中提出的基于WebSocket協議的實時消息系統是一種輕量的、平臺無關的設計方案?；诖朔桨?，企業可快速向現有平臺集成消息推送等功能。

參考文獻：

[1] 張積存，宋雪萍，費繼友，等.基于車輛信息的大數據分析系統設計與實現 [J].計算機應用與軟件，2023，40（1）：11-16+37.

[2] 蘇玉釗，孫恩昌，楊睿哲，等.面向Kafka共識消息傳輸的負載均衡算法 [J].高技術通訊，2023，33（1）：42-49.

[3] 孫海勇，金鴻宇，富宏亞，等.面向實時智能控制的數據采集平臺及其應用研究[J].航空制造技術，2022，65（22）：87-93.

[4] 陳樂，余粟，王盟.基于分布式集群的高可用日志分析系統的設計 [J].中國電子科學研究院學報，2020，15（5）：420-426.

[5] AMOZARRAIN U，LARREA M. Using publish/subscribe for message routing in mobile environments [J].Wireless Networks，2023，29：1831-1842.

[6] 李鵬程，張文勝，郭棟，等.基于物聯網通信協議的車輛信息系統開發 [J].計算機工程與設計，2022，43（3）：646-653.

[7] 高志輝，秦琦，段暕，等.基于實時Web技術的車間監測系統設計與實現 [J].計算機應用，2023，43（S1）：201-206.

[8] 胡喜明，胡淼.基于響應式的服務推送框架設計 [J].計算機工程與設計，2021，42（8）：2137-2143.

[9] 沈海波，陳強，陳勇昌.基于OAuth 2.0擴展的客戶端認證方案 [J].計算機工程與設計，2017，38（2）：350-354.

[10] 朱俊.基于規則分組的DFA正則表達式匹配算法 [J].湖南工程學院學報：自然科學版，2021，31（2）：49-53.

作者簡介：邢如意（1982—），男，漢族，江蘇徐州人，講師，碩士，研究方向：云計算、微服務架構。