前言

先簡單說下本次的主題，由于我最近做的是物聯(lián)網(wǎng)相關(guān)的開發(fā)工作，其中就不免會遇到和設(shè)備的交互。

創(chuàng)新互聯(lián)建站主要從事成都做網(wǎng)站、成都網(wǎng)站建設(shè)、網(wǎng)頁設(shè)計(jì)、企業(yè)做網(wǎng)站、公司建網(wǎng)站等業(yè)務(wù)。立足成都服務(wù)大武口,十載網(wǎng)站建設(shè)經(jīng)驗(yàn),價(jià)格優(yōu)惠、服務(wù)專業(yè),歡迎來電咨詢建站服務(wù):18982081108

最主要的工作就是要有一個系統(tǒng)來支持設(shè)備的接入、向設(shè)備推送消息；同時還得滿足大量設(shè)備接入的需求。

所以本次分享的內(nèi)容不但可以滿足物聯(lián)網(wǎng)領(lǐng)域同時還支持以下場景：

• 基于 WEB 的聊天系統(tǒng)（點(diǎn)對點(diǎn)、群聊）。
• WEB 應(yīng)用中需求服務(wù)端推送的場景。
• 基于 SDK 的消息推送平臺。

技術(shù)選型

要滿足大量的連接數(shù)、同時支持雙全工通信，并且性能也得有保障。

在 Java 技術(shù)棧中進(jìn)行選型首先自然是排除掉了傳統(tǒng) IO。

那就只有選 NIO 了，在這個層面其實(shí)選擇也不多，考慮到社區(qū)、資料維護(hù)等方面最終選擇了 Netty。

最終的架構(gòu)圖如下：

現(xiàn)在看著蒙沒關(guān)系，下文一一介紹。

協(xié)議解析

既然是一個消息系統(tǒng)，那自然得和客戶端定義好雙方的協(xié)議格式。

常見和簡單的是 HTTP 協(xié)議，但我們的需求中有一項(xiàng)需要是雙全工的交互方式，同時 HTTP 更多的是服務(wù)于瀏覽器。我們需要的是一個更加精簡的協(xié)議，減少許多不必要的數(shù)據(jù)傳輸。

因此我覺得最好是在滿足業(yè)務(wù)需求的情況下定制自己的私有協(xié)議，在我這個場景下其實(shí)有標(biāo)準(zhǔn)的物聯(lián)網(wǎng)協(xié)議。

如果是其他場景可以借鑒現(xiàn)在流行的 RPC 框架定制私有協(xié)議，使得雙方通信更加高效。

不過根據(jù)這段時間的經(jīng)驗(yàn)來看，不管是哪種方式都得在協(xié)議中預(yù)留安全相關(guān)的位置。

協(xié)議相關(guān)的內(nèi)容就不過討論了，更多介紹具體的應(yīng)用。

簡單實(shí)現(xiàn)

首先考慮如何實(shí)現(xiàn)功能，再來思考百萬連接的情況。

注冊鑒權(quán)

在做真正的消息上、下行之前首先要考慮的就是鑒權(quán)問題。

就像你使用微信一樣，第一步怎么也得是登錄吧，不能無論是誰都可以直接連接到平臺。

所以第一步得是注冊才行。

如上面架構(gòu)圖中的 注冊/鑒權(quán) 模塊。通常來說都需要客戶端通過 HTTP 請求傳遞一個唯一標(biāo)識，后臺鑒權(quán)通過之后會響應(yīng)一個 token，并將這個 token 和客戶端的關(guān)系維護(hù)到 redis 或者是 DB 中。

客戶端將這個 token 也保存到本地，今后的每一次請求都得帶上這個 token。一旦這個 token 過期，客戶端需要再次請求獲取 token。

鑒權(quán)通過之后客戶端會直接通過 TCP長連接到圖中的 push-server 模塊。

這個模塊就是真正處理消息的上、下行。

保存通道關(guān)系

在連接接入之后，真正處理業(yè)務(wù)之前需要將當(dāng)前的客戶端和 Channel 的關(guān)系維護(hù)起來。

假設(shè)客戶端的唯一標(biāo)識是手機(jī)號碼，那就需要把手機(jī)號碼和當(dāng)前的 Channel 維護(hù)到一個 Map 中。

這點(diǎn)和之前 SpringBoot 整合長連接心跳機(jī)制 類似。

同時為了可以通過 Channel 獲取到客戶端唯一標(biāo)識（手機(jī)號碼），還需要在 Channel 中設(shè)置對應(yīng)的屬性：

public

static

void
 putClientId
(
Channel
 channel
,

String
 clientId
)

{
 channel
.
attr
(
CLIENT_ID
).
set
(
clientId
);
}

獲取時手機(jī)號碼時：

public

static

String
 getClientId
(
Channel
 channel
)

{

return

(
String
)
getAttribute
(
channel
,
 CLIENT_ID
);
}

這樣當(dāng)我們客戶端下線的時便可以記錄相關(guān)日志：

String
 telNo 
=

NettyAttrUtil
.
getClientId
(
ctx
.
channel
());
NettySocketHolder
.
remove
(
telNo
);
log
.
info
(
"客戶端下線，TelNo="

+
 telNo
);

這里有一點(diǎn)需要注意：存放客戶端與 Channel 關(guān)系的 Map 最好是預(yù)設(shè)好大小（避免經(jīng)常擴(kuò)容），因?yàn)樗鼘⑹鞘褂米顬轭l繁同時也是占用內(nèi)存最大的一個對象。

消息上行

接下來則是真正的業(yè)務(wù)數(shù)據(jù)上傳，通常來說第一步是需要判斷上傳消息輸入什么業(yè)務(wù)類型。

在聊天場景中，有可能上傳的是文本、圖片、視頻等內(nèi)容。

所以我們得進(jìn)行區(qū)分，來做不同的處理；這就和客戶端協(xié)商的協(xié)議有關(guān)了。

• 可以利用消息頭中的某個字段進(jìn)行區(qū)分。
• 更簡單的就是一個 JSON 消息，拿出一個字段用于區(qū)分不同消息。

不管是哪種只有可以區(qū)分出來即可。

消息解析與業(yè)務(wù)解耦

消息可以解析之后便是處理業(yè)務(wù)，比如可以是寫入數(shù)據(jù)庫、調(diào)用其他接口等。

我們都知道在 Netty 中處理消息一般是在 channelRead() 方法中。

在這里可以解析消息，區(qū)分類型。

但如果我們的業(yè)務(wù)邏輯也寫在里面，那這里的內(nèi)容將是巨多無比。

甚至我們分為好幾個開發(fā)來處理不同的業(yè)務(wù)，這樣將會出現(xiàn)許多沖突、難以維護(hù)等問題。

所以非常有必要將消息解析與業(yè)務(wù)處理完全分離開來。

這時面向接口編程就發(fā)揮作用了。

這里的核心代碼和 「造個輪子」——cicada(輕量級 WEB 框架) 是一致的。

都是先定義一個接口用于處理業(yè)務(wù)邏輯，然后在解析消息之后通過反射創(chuàng)建具體的對象執(zhí)行其中的 處理函數(shù)即可。

這樣不同的業(yè)務(wù)、不同的開發(fā)人員只需要實(shí)現(xiàn)這個接口同時實(shí)現(xiàn)自己的業(yè)務(wù)邏輯即可。

偽代碼如下：

上行還有一點(diǎn)需要注意；由于是基于長連接，所以客戶端需要定期發(fā)送心跳包用于維護(hù)本次連接。同時服務(wù)端也會有相應(yīng)的檢查，N 個時間間隔沒有收到消息之后將會主動斷開連接節(jié)省資源。

這點(diǎn)使用一個 IdleStateHandler 就可實(shí)現(xiàn)，更多內(nèi)容可以查看 Netty(一) SpringBoot 整合長連接心跳機(jī)制。

消息下行

有了上行自然也有下行。比如在聊天的場景中，有兩個客戶端連上了 push-server,他們直接需要點(diǎn)對點(diǎn)通信。

這時的流程是：

• A 將消息發(fā)送給服務(wù)器。
• 服務(wù)器收到消息之后，得知消息是要發(fā)送給 B，需要在內(nèi)存中找到 B 的 Channel。
• 通過 B 的 Channel 將 A 的消息轉(zhuǎn)發(fā)下去。

這就是一個下行的流程。

甚至管理員需要給所有在線用戶發(fā)送系統(tǒng)通知也是類似：

遍歷保存通道關(guān)系的 Map，挨個發(fā)送消息即可。這也是之前需要存放到 Map 中的主要原因。

偽代碼如下：

分布式方案

單機(jī)版的實(shí)現(xiàn)了，現(xiàn)在著重講講如何實(shí)現(xiàn)百萬連接。

百萬連接其實(shí)只是一個形容詞，更多的是想表達(dá)如何來實(shí)現(xiàn)一個分布式的方案，可以靈活的水平拓展從而能支持更多的連接。

再做這個事前首先得搞清楚我們單機(jī)版的能支持多少連接。影響這個的因素就比較多了。

• 服務(wù)器自身配置。內(nèi)存、CPU、網(wǎng)卡、Linux 支持的最大文件打開數(shù)等。
• 應(yīng)用自身配置，因?yàn)?Netty 本身需要依賴于堆外內(nèi)存，但是 JVM 本身也是需要占用一部分內(nèi)存的，比如存放通道關(guān)系的大 Map。這點(diǎn)需要結(jié)合自身情況進(jìn)行調(diào)整。

結(jié)合以上的情況可以測試出單個節(jié)點(diǎn)能支持的最大連接數(shù)。

單機(jī)無論怎么優(yōu)化都是有上限的，這也是分布式主要解決的問題。

架構(gòu)介紹

在將具體實(shí)現(xiàn)之前首先得講講上文貼出的整體架構(gòu)圖。

先從左邊開始。

上文提到的 注冊鑒權(quán) 模塊也是集群部署的，通過前置的 Nginx 進(jìn)行負(fù)載。之前也提過了它主要的目的是來做鑒權(quán)并返回一個 token 給客戶端。

但是 push-server 集群之后它又多了一個作用。那就是得返回一臺可供當(dāng)前客戶端使用的 push-server。

右側(cè)的 平臺 一般指管理平臺，它可以查看當(dāng)前的實(shí)時在線數(shù)、給指定客戶端推送消息等。

推送消息則需要經(jīng)過一個推送路由（ push-server）找到真正的推送節(jié)點(diǎn)。

其余的中間件如：Redis、Zookeeper、Kafka、MySQL 都是為了這些功能所準(zhǔn)備的，具體看下面的實(shí)現(xiàn)。

注冊發(fā)現(xiàn)

首先第一個問題則是 注冊發(fā)現(xiàn)， push-server 變?yōu)槎嗯_之后如何給客戶端選擇一臺可用的節(jié)點(diǎn)是第一個需要解決的。

這塊的內(nèi)容其實(shí)已經(jīng)在 分布式(一) 搞定服務(wù)注冊與發(fā)現(xiàn) 中詳細(xì)講過了。

所有的 push-server 在啟動時候需要將自身的信息注冊到 Zookeeper 中。

注冊鑒權(quán) 模塊會訂閱 Zookeeper 中的節(jié)點(diǎn)，從而可以獲取最新的服務(wù)列表。結(jié)構(gòu)如下：

以下是一些偽代碼：

應(yīng)用啟動注冊 Zookeeper。

對于 注冊鑒權(quán)模塊來說只需要訂閱這個 Zookeeper 節(jié)點(diǎn)：

路由策略

既然能獲取到所有的服務(wù)列表，那如何選擇一臺剛好合適的 push-server 給客戶端使用呢？

這個過程重點(diǎn)要考慮以下幾點(diǎn)：

• 盡量保證各個節(jié)點(diǎn)的連接均勻。
• 增刪節(jié)點(diǎn)是否要做 Rebalance。

首先保證均衡有以下幾種算法：

• 輪詢。挨個將各個節(jié)點(diǎn)分配給客戶端。但會出現(xiàn)新增節(jié)點(diǎn)分配不均勻的情況。
• Hash 取模的方式。類似于 HashMap，但也會出現(xiàn)輪詢的問題。當(dāng)然也可以像 HashMap 那樣做一次 Rebalance，讓所有的客戶端重新連接。不過這樣會導(dǎo)致所有的連接出現(xiàn)中斷重連，代價(jià)有點(diǎn)大。
• 由于 Hash 取模方式的問題帶來了 一致性Hash算法，但依然會有一部分的客戶端需要 Rebalance。
• 權(quán)重?？梢允謩诱{(diào)整各個節(jié)點(diǎn)的負(fù)載情況，甚至可以做成自動的，基于監(jiān)控當(dāng)某些節(jié)點(diǎn)負(fù)載較高就自動調(diào)低權(quán)重，負(fù)載較低的可以提高權(quán)重。

還有一個問題是：

當(dāng)我們在重啟部分應(yīng)用進(jìn)行升級時，在該節(jié)點(diǎn)上的客戶端怎么處理？

由于我們有心跳機(jī)制，當(dāng)心跳不通之后就可以認(rèn)為該節(jié)點(diǎn)出現(xiàn)問題了。那就得重新請求 注冊鑒權(quán)模塊獲取一個可用的節(jié)點(diǎn)。在弱網(wǎng)情況下同樣適用。

如果這時客戶端正在發(fā)送消息，則需要將消息保存到本地等待獲取到新的節(jié)點(diǎn)之后再次發(fā)送。

有狀態(tài)連接

在這樣的場景中不像是 HTTP 那樣是無狀態(tài)的，我們得明確的知道各個客戶端和連接的關(guān)系。

在上文的單機(jī)版中我們將這個關(guān)系保存到本地的緩存中，但在分布式環(huán)境中顯然行不通了。

比如在平臺向客戶端推送消息的時候，它得首先知道這個客戶端的通道保存在哪臺節(jié)點(diǎn)上。

借助我們以前的經(jīng)驗(yàn)，這樣的問題自然得引入一個第三方中間件用來存放這個關(guān)系。

也就是架構(gòu)圖中的存放 路由關(guān)系的Redis，在客戶端接入 push-server 時需要將當(dāng)前客戶端唯一標(biāo)識和服務(wù)節(jié)點(diǎn)的 ip+port 存進(jìn) Redis。

同時在客戶端下線時候得在 Redis 中刪掉這個連接關(guān)系。

這樣在理想情況下各個節(jié)點(diǎn)內(nèi)存中的 map 關(guān)系加起來應(yīng)該正好等于 Redis 中的數(shù)據(jù)。

偽代碼如下：

這里存放路由關(guān)系的時候會有并發(fā)問題，最好是換為一個 lua 腳本。

推送路由

設(shè)想這樣一個場景：管理員需要給最近注冊的客戶端推送一個系統(tǒng)消息會怎么做？

結(jié)合架構(gòu)圖

假設(shè)這批客戶端有 10W 個，首先我們需要將這批號碼通過 平臺下的 Nginx 下發(fā)到一個推送路由中。

為了提高效率甚至可以將這批號碼再次分散到每個 push-route 中。

拿到具體號碼之后再根據(jù)號碼的數(shù)量啟動多線程的方式去之前的路由 Redis 中獲取客戶端所對應(yīng)的 push-server。

再通過 HTTP 的方式調(diào)用 push-server 進(jìn)行真正的消息下發(fā)（Netty 也很好的支持 HTTP 協(xié)議）。

推送成功之后需要將結(jié)果更新到數(shù)據(jù)庫中，不在線的客戶端可以根據(jù)業(yè)務(wù)再次推送等。

消息流轉(zhuǎn)

也許有些場景對于客戶端上行的消息非常看重，需要做持久化，并且消息量非常大。

在 push-sever 做業(yè)務(wù)顯然不合適，這時完全可以選擇 Kafka 來解耦。

將所有上行的數(shù)據(jù)直接往 Kafka 里丟后就不管了。

再由消費(fèi)程序?qū)?shù)據(jù)取出寫入數(shù)據(jù)庫中即可。

其實(shí)這塊內(nèi)容也很值得討論，可以先看這篇了解下：強(qiáng)如 Disruptor 也發(fā)生內(nèi)存溢出？

后續(xù)談到 Kafka 再做詳細(xì)介紹。

分布式問題

分布式解決了性能問題但卻帶來了其他麻煩。

應(yīng)用監(jiān)控

比如如何知道線上幾十個 push-server 節(jié)點(diǎn)的健康狀況？

這時就得監(jiān)控系統(tǒng)發(fā)揮作用了，我們需要知道各個節(jié)點(diǎn)當(dāng)前的內(nèi)存使用情況、GC。

以及操作系統(tǒng)本身的內(nèi)存使用，畢竟 Netty 大量使用了堆外內(nèi)存。

同時需要監(jiān)控各個節(jié)點(diǎn)當(dāng)前的在線數(shù)，以及 Redis 中的在線數(shù)。理論上這兩個數(shù)應(yīng)該是相等的。

這樣也可以知道系統(tǒng)的使用情況，可以靈活的維護(hù)這些節(jié)點(diǎn)數(shù)量。

日志處理

日志記錄也變得異常重要了，比如哪天反饋有個客戶端一直連不上，你得知道問題出在哪里。

最好是給每次請求都加上一個 traceID 記錄日志，這樣就可以通過這個日志在各個節(jié)點(diǎn)中查看到底是卡在了哪里。

以及 ELK 這些工具都得用起來才行。

總結(jié)

本次是結(jié)合我日常經(jīng)驗(yàn)得出的，有些坑可能在工作中并沒有踩到，所有還會有一些遺漏的地方。

就目前來看想做一個穩(wěn)定的推送系統(tǒng)其實(shí)是比較麻煩的，其中涉及到的點(diǎn)非常多，只有真正做過之后才會知道。