收稿日期：2023-06-30

DOI：10.19850/j.cnki.2096-4706.2023.21.007

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

關鍵詞：發(fā)布訂閱；實時消息系統(tǒng)；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是當前使用廣泛和優(yōu)秀的分布式消息平臺，其提供了基于發(fā)布訂閱機制的消息系統(tǒng)，可以實現消息生產者生產消息發(fā)送到消息系統(tǒng)，消息系統(tǒng)投遞給消費者；提供了大容量存儲系統(tǒng)，支持大數據量存儲和歷史數據存儲；提供了實時數據管道，支持建立面向MySQL等數據源的實時數據管道，繼而為其他系統(tǒng)提供數據服務；提供了流計算能力，支持與Spark、Flink等分布式計算引擎結合，構建實時流應用，實時處理大規(guī)模數據。

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

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

1" 技術選型

1.1" 消息傳遞模式選型

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

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

1.2" 消息傳遞協議選型

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

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

1.3" 服務端實現框架選型

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

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

2" 系統(tǒng)設計

2.1" 架構設計

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

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

2.2" 詳細設計

2.2.1" 全局ID設計

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

為壓縮全局ID的長度且保持可讀性，文中參照Snowflake算法并進行了修改，將全局ID設計為34位長度的字符。圖2展示了全局ID字符的結構，分別包括時間（14位）、用戶編號（10位）、序列號（10位）。其中時間部分格式為年月日時分秒、序列號部分使用Redis生成?；诖朔椒ㄉ扇治ㄒ籌D使得消息鑒別更具業(yè)務屬性，便于數據調試。

2.2.2" 系統(tǒng)組件調用流程設計

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

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

2.2.3" 訪問令牌設計

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

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

2.2.4" 輸入敏感詞過濾

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

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

3" 系統(tǒng)實現與測試

3.1" 消息廣播功能實現

使用Socket.IO與Redis實現消息廣播功能，使用NodeJS中的http模塊編寫Socket.IO服務端，創(chuàng)建/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服務，監(jiān)聽Redis的廣播消息，根據頻道的不同將消息分發(fā)給客戶端，關鍵代碼如下：

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

3.2" 測試結果分析

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

4" 結" 論

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

參考文獻：

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

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

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

[4] 陳樂，余粟，王盟.基于分布式集群的高可用日志分析系統(tǒng)的設計 [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] 李鵬程，張文勝，郭棟，等.基于物聯網通信協議的車輛信息系統(tǒng)開發(fā) [J].計算機工程與設計，2022，43（3）：646-653.

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

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

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

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

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