一、引言

要實(shí)現(xiàn)一整套能用于大用戶量、高并發(fā)場(chǎng)景下的IM群聊，技術(shù)難度遠(yuǎn)超IM系統(tǒng)中的其它功能，原因在于：IM群聊消息的實(shí)時(shí)寫擴(kuò)散特性帶來(lái)了一系列技術(shù)難題。

舉個(gè)例子：如一個(gè)2000人群里，一條普通消息的發(fā)出問(wèn)題，將瞬間寫擴(kuò)散為2000條消息的接收問(wèn)題，如何保證這些消息的及時(shí)、有序、高效地送達(dá)，涉及到的技術(shù)問(wèn)題點(diǎn)實(shí)在太多，更別說(shuō)個(gè)別場(chǎng)景下萬(wàn)人大群里的炸群消息難題了更別說(shuō)個(gè)別場(chǎng)景下萬(wàn)人大群里的炸群消息難題了。

這也是為什么一般中大型IM系統(tǒng)中，都會(huì)將群聊單獨(dú)拎出來(lái)考慮架構(gòu)的設(shè)計(jì)，單獨(dú)有針對(duì)性地進(jìn)行架構(gòu)優(yōu)化，從而降低整個(gè)系統(tǒng)的設(shè)計(jì)難度。

本文將分享的是一套生產(chǎn)環(huán)境下的IM群聊消息系統(tǒng)的高可用、易伸縮、高并發(fā)架構(gòu)設(shè)計(jì)實(shí)踐，屬于原創(chuàng)第一手資料，內(nèi)容較專業(yè)，適合有一定IM架構(gòu)經(jīng)驗(yàn)的后端程序員閱讀。

二、群聊技術(shù)文章

《IM群聊消息究竟是存1份(即擴(kuò)散讀)還是存多份(即擴(kuò)散寫)？》
《IM群聊消息的已讀回執(zhí)功能該怎么實(shí)現(xiàn)？》
《關(guān)于IM即時(shí)通訊群聊消息的亂序問(wèn)題討論》
《現(xiàn)代IM系統(tǒng)中聊天消息的同步和存儲(chǔ)方案探討》
《移動(dòng)端IM中大規(guī)模群消息的推送如何保證效率、實(shí)時(shí)性？》
《微信后臺(tái)團(tuán)隊(duì)：微信后臺(tái)異步消息隊(duì)列的優(yōu)化升級(jí)實(shí)踐分享》
《IM群聊消息如此復(fù)雜，如何保證不丟不重？》
《IM單聊和群聊中的在線狀態(tài)同步應(yīng)該用“推”還是“拉”？》
《如何保證IM實(shí)時(shí)消息的“時(shí)序性”與“一致性”？》
《快速裂變：見(jiàn)證微信強(qiáng)大后臺(tái)架構(gòu)從0到1的演進(jìn)歷程（一）》

三、萬(wàn)事開頭難：初始的極簡(jiǎn)實(shí)現(xiàn)

所謂的群聊消息系統(tǒng)，就是一種多對(duì)多群體聊天方式，譬如直播房間內(nèi)的聊天室對(duì)應(yīng)的服務(wù)器端就是一個(gè)群聊消息系統(tǒng)。

2017年9月初，我們初步實(shí)現(xiàn)了一套極簡(jiǎn)的群聊消息系統(tǒng)，其大致架構(gòu)如下：

系統(tǒng)名詞解釋：
?

• 1）Client :?消息發(fā)布者【或者叫做服務(wù)端群聊消息系統(tǒng)調(diào)用者】，publisher；
• 2）Proxy :?系統(tǒng)代理，對(duì)外統(tǒng)一接口，收集Client發(fā)來(lái)的消息轉(zhuǎn)發(fā)給Broker；
• 3）Broker ：系統(tǒng)消息轉(zhuǎn)發(fā)Server，Broker 會(huì)根據(jù) Gateway Message 組織一個(gè) RoomGatewayList【key為RoomID，value為 Gateway IP：Port 地址列表】，然后把 Proxy 發(fā)來(lái)的消息轉(zhuǎn)發(fā)到 Room 中所有成員登錄的所有 Gateway；
• 4）Router ：用戶登錄消息轉(zhuǎn)發(fā)者，把Gateway轉(zhuǎn)發(fā)來(lái)的用戶登入登出消息轉(zhuǎn)發(fā)給所有的Broker；
• 5）Gateway ：所有服務(wù)端的入口，接收合法客戶端的連接，并把客戶端的登錄登出消息通過(guò)Router轉(zhuǎn)發(fā)給所有的Broker；
• 6）Room Message :?Room聊天消息；
• 7）Gateway Message :?Room內(nèi)某成員 登錄 或者 登出 某Gateway消息，包含用戶UIN/RoomID/Gateway地址{IP：Port}等消息。

當(dāng)一個(gè) Room 中多個(gè) Client 連接一個(gè) Gateway 的時(shí)候，Broker只會(huì)根據(jù) RoomID 把房間內(nèi)的消息轉(zhuǎn)發(fā)一次給這個(gè)Gateway，由Gateway再把消息復(fù)制多份分別發(fā)送給連接這個(gè) Gateway 的 Room 中的所有用戶的客戶端。

這套系統(tǒng)有如下特點(diǎn)：
?

• 1）系統(tǒng)只轉(zhuǎn)發(fā)房間內(nèi)的聊天消息，每個(gè)節(jié)點(diǎn)收到后立即轉(zhuǎn)發(fā)出去，不存儲(chǔ)任何房間內(nèi)的聊天消息，不考慮消息丟失以及消息重復(fù)的問(wèn)題；
• 2）系統(tǒng)固定地由一個(gè)Proxy、三個(gè)Broker和一個(gè)Router構(gòu)成；
• 3）Proxy接收后端發(fā)送來(lái)的房間消息，然后按照一定的負(fù)載均衡算法把消息發(fā)往某個(gè)Broker，Broker則把消息發(fā)送到所有與Room有關(guān)系的接口機(jī)Gateway；
• 4）Router接收Gateway轉(zhuǎn)發(fā)來(lái)的某個(gè)Room內(nèi)某成員在這個(gè)Gateway的登出或者登錄消息，然后把消息發(fā)送到所有Broker；
• 5）Broker收到Router轉(zhuǎn)發(fā)來(lái)的Gateway消息后，更新（添加或者刪除）與某Room相關(guān)的Gateway集合記錄；
• 6）整個(gè)系統(tǒng)的通信鏈路采用UDP通信方式。

從以上特點(diǎn)，整個(gè)消息系統(tǒng)足夠簡(jiǎn)單，沒(méi)有考慮擴(kuò)縮容問(wèn)題，當(dāng)系統(tǒng)負(fù)載到達(dá)極限的時(shí)候，就重新再部署一套系統(tǒng)以應(yīng)對(duì)后端client的消息壓力。

這種處理方式本質(zhì)是把系統(tǒng)的擴(kuò)容能力甩鍋給了后端Client以及前端Gateway：每次擴(kuò)容一個(gè)系統(tǒng)，所有Client需要在本地配置文件中添加一個(gè)Proxy地址然后全部重啟，所有Gateway則需要再本地配置文件添加一個(gè)Router地址然后全部重啟。

這種“幸福我一人，辛苦千萬(wàn)家”的擴(kuò)容應(yīng)對(duì)方式，必然導(dǎo)致公司內(nèi)部這套系統(tǒng)的使用者怨聲載道，下一階段的升級(jí)就是必然的了。

四、進(jìn)一步重點(diǎn)設(shè)計(jì)：“可擴(kuò)展性”

?

4.1、基本思路

大道之行也，天下為公，不同的系統(tǒng)有不同的構(gòu)架，相同的系統(tǒng)總有類似的實(shí)現(xiàn)。類似于數(shù)據(jù)庫(kù)的分庫(kù)分表【關(guān)于分庫(kù)分表，目前看到的最好的文章是《一種支持自由規(guī)劃無(wú)須數(shù)據(jù)遷移和修改路由代碼的Replicaing擴(kuò)容方案》】，其擴(kuò)展實(shí)現(xiàn)核心思想是分Partition分Replica，但各Replica之間還區(qū)分leader（leader-follower，只有l(wèi)eader可接受寫請(qǐng)求）和non-leader（所有replica均可接收寫請(qǐng)求）兩種機(jī)制。

從數(shù)據(jù)角度來(lái)看，這套系統(tǒng)接收兩種消息：Room Message（房間聊天消息）和Gateway Message（用戶登錄消息）。兩種消息的交匯之地就是Broker，所以應(yīng)對(duì)擴(kuò)展的緊要地方就是Broker，Broker的每個(gè)Partition采用non-leader機(jī)制，各replica均可接收Gateway Message消息寫請(qǐng)求和Room Message轉(zhuǎn)發(fā)請(qǐng)求。

首先，當(dāng)Room Message量加大時(shí)可以對(duì)Proxy進(jìn)行水平擴(kuò)展，多部署Proxy即可因應(yīng)Room Message的流量。

其次，當(dāng)Gateway Message量加大時(shí)可以對(duì)Router進(jìn)行水平擴(kuò)展，多部署Router即可因應(yīng)Gateway Message的流量。

最后，兩種消息的交匯之地Broker如何擴(kuò)展呢？可以把若干Broker Replica組成一個(gè)Partition，因?yàn)镚ateway Message是在一個(gè)Partition內(nèi)廣播的，所有Broker Replica都會(huì)有相同的RoomGatewayList 數(shù)據(jù)，因此當(dāng)Gateway Message增加時(shí)擴(kuò)容Partition即可。當(dāng)Room Message量增加時(shí)，水平擴(kuò)容Partition內(nèi)的Broker Replica即可，因?yàn)镽oom Message只會(huì)發(fā)送到Partition內(nèi)某個(gè)Replica上。

從個(gè)人經(jīng)驗(yàn)來(lái)看，Room ID的增長(zhǎng)以及Room內(nèi)成員的增加量在一段時(shí)間內(nèi)可以認(rèn)為是直線增加，而Room Message可能會(huì)以指數(shù)級(jí)增長(zhǎng)，所以若設(shè)計(jì)得當(dāng)則Partition擴(kuò)容的概率很小，而Partition內(nèi)Replica水平增長(zhǎng)的概率幾乎是100%。

不管是Partition級(jí)別的水平擴(kuò)容還是Partition Replica級(jí)別的水平擴(kuò)容，不可能像系統(tǒng)極簡(jiǎn)版本那樣每次擴(kuò)容后都需要Client或者Gateway去更新配置文件然后重啟，因應(yīng)之道就是可用zookeeper充當(dāng)角色的Registriy。通過(guò)這個(gè)zookeeper注冊(cè)中心，相關(guān)角色擴(kuò)容的時(shí)候在Registry注冊(cè)后，與之相關(guān)的其他模塊得到通知即可獲取其地址等信息。采用zookeeper作為Registry的時(shí)候，所以程序?qū)崿F(xiàn)的時(shí)候采用實(shí)時(shí)watch和定時(shí)輪詢的策略保證數(shù)據(jù)可靠性，因?yàn)橐坏┚W(wǎng)絡(luò)有任何的抖動(dòng)，zk就會(huì)認(rèn)為客戶端已經(jīng)宕機(jī)把鏈接關(guān)閉。

分析完畢，與之相對(duì)的架構(gòu)圖如下：

以下各分章節(jié)將描述各個(gè)模塊詳細(xì)流程。

4.2、Client

Client詳細(xì)流程如下：
?

• 1）從配置文件加載Registry地址；
• 2）從Registy上Proxy注冊(cè)路徑/pubsub/proxy下獲取所有的Proxy，依據(jù)各個(gè)Proxy ID大小順序遞增組成一個(gè)ProxyArray；
• 3）啟動(dòng)一個(gè)線程實(shí)時(shí)關(guān)注Registry路徑/pubsub/proxy，以獲取Proxy的動(dòng)態(tài)變化，及時(shí)更新ProxyArray；
• 4）啟動(dòng)一個(gè)線程定時(shí)輪詢獲取Registry路徑/pubsub/proxy下各個(gè)Proxy實(shí)例，作為關(guān)注策略的補(bǔ)充，以期本地ProxyArray內(nèi)各個(gè)Proxy成員與Registry上的各個(gè)Proxy保持一致；定時(shí)給各個(gè)Proxy發(fā)送心跳，異步獲取心跳回包；定時(shí)清除ProxyArray中心跳超時(shí)的Proxy成員；
• 5）發(fā)送消息的時(shí)候采用snowflake算法給每個(gè)消息分配一個(gè)MessageID，然后采用相關(guān)負(fù)載均衡算法把消息轉(zhuǎn)發(fā)給某個(gè)Proxy。

?

4.3、Proxy

Proxy詳細(xì)流程如下：
?

• 1）讀取配置文件，獲取Registry地址；
• 2）把自身信息注冊(cè)到Registry路徑/pubsub/proxy下，把Registry返回的ReplicaID作為自身ID；
• 3）從Registry路徑/pubsub/broker/partition(x)下獲取每個(gè)Broker Partition的各個(gè)replica；
• 4）從Registry路徑/pubsub/broker/partition_num獲取當(dāng)前有效的Broker Partition Number；

?

• 5）啟動(dòng)一個(gè)線程關(guān)注Registry上的Broker路徑/pubsub/broker，以實(shí)時(shí)獲取以下信息：
• ? ? {Broker Partition Number}
• ? ???- 新的Broker Partition（此時(shí)發(fā)生了擴(kuò)容）；
• ? ???- Broker Partition內(nèi)新的broker replica（Partition內(nèi)發(fā)生了replica擴(kuò)容）；
• ? ???- Broker Parition內(nèi)某replica掛掉的信息；

?

• 6）定時(shí)向各個(gè)Broker Partition replica發(fā)送心跳，異步等待Broker返回的心跳響應(yīng)包，以探測(cè)其活性，以保證不向超時(shí)的replica轉(zhuǎn)發(fā)Room Message；
• 7）啟動(dòng)一個(gè)線程定時(shí)讀取Registry上的Broker路徑/pubsub/broker下各個(gè)子節(jié)點(diǎn)的值，以定時(shí)輪詢的策略觀察Broker Partition Number變動(dòng)，以及各Partition的變動(dòng)情況，作為實(shí)時(shí)策略的補(bǔ)充；同時(shí)定時(shí)檢查心跳包超時(shí)的Broker，從有效的BrokerList中刪除；
• 8）依據(jù)規(guī)則【BrokerPartitionID = RoomID % BrokerPartitionNum， BrokerReplicaID = RoomID % BrokerPartitionReplicaNum】向某個(gè)Partition的replica轉(zhuǎn)發(fā)Room Message，收到Client的Heatbeat包時(shí)要及時(shí)給予響應(yīng)。

之所以把Room Message和Heartbeat Message放在一個(gè)線程處理，是為了防止進(jìn)程假死這種情況。

當(dāng)/pubsub/broker/partition_num的值發(fā)生改變的時(shí)候(譬如值改為4)，意味著Router Partition進(jìn)行了擴(kuò)展，Proxy要及時(shí)獲取新Partition路徑（如/pubsub/broker/Partition2和/pubsub/broker/Partition3）下的實(shí)例，并關(guān)注這些路徑，獲取新Partition下的實(shí)例。

之所以Proxy在獲取Registry下所有當(dāng)前的Broker實(shí)例信息后再注冊(cè)自身信息，是因?yàn)榇藭r(shí)它才具有轉(zhuǎn)發(fā)消息的資格。

Proxy轉(zhuǎn)發(fā)某個(gè)Room消息時(shí)候，只發(fā)送給處于Running狀態(tài)的Broker。為Broker Partition內(nèi)所有replica依據(jù)Registry給其分配的replicaID進(jìn)行遞增排序，組成一個(gè)Broker Partition Replica Array，規(guī)則中BrokerPartitionReplicaNum為Array的size，而BrokerReplicaID為replica在Array中的下標(biāo)。

4.4、Pipeline

收到的 Room Message 需要做三部工作：收取 Room Message、消息協(xié)議轉(zhuǎn)換和向 Broker 發(fā)送消息。

初始系統(tǒng)這三步流程如果均放在一個(gè)線程內(nèi)處理，proxy 的整體吞吐率只有 50 000 Msg/s。

最后的實(shí)現(xiàn)方式是按照消息處理的三個(gè)步驟以 pipeline 方式做如下流程處理：
?

• 1）啟動(dòng) 1 個(gè)消息接收線程和 N【N == Broker Parition 數(shù)目】個(gè)多寫一讀形式的無(wú)鎖隊(duì)列【稱之為消息協(xié)議轉(zhuǎn)換隊(duì)列】，消息接收線程分別啟動(dòng)一個(gè) epoll 循環(huán)流程收取消息，然后把消息以相應(yīng)的 hash 算法【隊(duì)列ID = UIN % N】寫入對(duì)應(yīng)的消息協(xié)議轉(zhuǎn)換隊(duì)列；
• 2）啟動(dòng) N 個(gè)線程 和 N * 3 個(gè)一寫一讀的無(wú)鎖隊(duì)列【稱之為消息發(fā)送隊(duì)列】，每個(gè)消息協(xié)議專家線程從消息協(xié)議轉(zhuǎn)換隊(duì)列接收到消息并進(jìn)行協(xié)議轉(zhuǎn)換后，根據(jù)相應(yīng)的 hash 算法【隊(duì)列ID = UIN % 3N】寫入消息發(fā)送隊(duì)列；
• 3）啟動(dòng) 3N 個(gè)消息發(fā)送線程，分別創(chuàng)建與之對(duì)應(yīng)的 Broker 的連接，每個(gè)線程單獨(dú)從對(duì)應(yīng)的某個(gè)消息發(fā)送隊(duì)列接收消息然后發(fā)送出去。

經(jīng)過(guò)以上流水線改造后，Proxy 的整體吞吐率可達(dá) 200 000 Msg/s。

關(guān)于 pipeline 自身的解釋，本文不做詳述，可以參考下圖：

4.5、大房間消息處理

每個(gè) Room 的人數(shù)不均，最簡(jiǎn)便的解決方法就是給不同人數(shù)量級(jí)的 Room 各搭建一套消息系統(tǒng)，不用修改任何代碼。

然所謂需求推動(dòng)架構(gòu)改進(jìn)，在系統(tǒng)迭代升級(jí)過(guò)程中遇到了這樣一個(gè)需求：業(yè)務(wù)方有一個(gè)全國(guó) Room，用于給所有在線用戶進(jìn)行消息推送。針對(duì)這個(gè)需求，不可能為了一個(gè)這樣的 Room 單獨(dú)搭建一套系統(tǒng)，況且這個(gè) Room 的消息量很少。

如果把這個(gè) Room 的消息直接發(fā)送給現(xiàn)有系統(tǒng)，它有可能影響其他 Room 的消息發(fā)送：消息系統(tǒng)是一個(gè)寫放大的系統(tǒng)，全國(guó) Room 內(nèi)有系統(tǒng)所有的在線用戶，每次發(fā)送都會(huì)卡頓其他 Room 的消息發(fā)送。

最終的解決方案是：使用類似于分區(qū)的方法，把這樣的大 Room 映射為 64 個(gè)虛擬 Room【稱之為 VRoom】。在 Room 號(hào)段分配業(yè)務(wù)線的配合下，給消息系統(tǒng)專門保留了一個(gè)號(hào)段，用于這種大 Room 的切分，在 Proxy 層依據(jù)一個(gè) hash 方法 【 VRoomID = UserID % 64】 把每個(gè) User 分配到相應(yīng)的 VRoom，其他模塊代碼不用修改即完成了大 Room 消息的路由。

4.6、Broker

Broker詳細(xì)流程如下：
?

• 1）Broker加載配置，獲取自身所在Partition的ID（假設(shè)為3）；
• 2）向Registry路徑/pubsub/broker/partition3注冊(cè)，設(shè)置其狀態(tài)為Init，注冊(cè)中心返回的ID作為自身的ID(replicaID)；
• 3）接收Router轉(zhuǎn)發(fā)來(lái)的Gateway Message，放入GatewayMessageQueue；
• 4）從Database加載數(shù)據(jù)，把自身所在的Broker Partition所應(yīng)該負(fù)責(zé)的 RoomGatewayList 數(shù)據(jù)加載進(jìn)來(lái)；
• 5）異步處理GatewayMessageQueue內(nèi)的Gateway Message，只處理滿足規(guī)則【PartitionID == RoomID % PartitionNum】的消息，把數(shù)據(jù)存入本地路由信息緩存；
• 6）修改Registry路徑/pubsub/broker/partition3下自身節(jié)點(diǎn)的狀態(tài)為Running；
• 7）啟動(dòng)線程實(shí)時(shí)關(guān)注Registry路徑/pubsub/broker/partition_num的值；
• 8）啟動(dòng)線程定時(shí)查詢Registry路徑/pubsub/broker/partition_num的值；
• 9）當(dāng)Registry路徑/pubsub/broker/partition_num的值發(fā)生改變的時(shí)候，依據(jù)規(guī)則【PartitionID == RoomID % PartitionNum】清洗本地路由信息緩存中每條數(shù)據(jù)；
• 10）接收Proxy發(fā)來(lái)的Room Message，依據(jù)RoomID從路由信息緩存中查找Room有成員登陸的所有Gateway，把消息轉(zhuǎn)發(fā)給這些Gateway。

注意Broker之所以先注冊(cè)然后再加載Database中的數(shù)據(jù)，是為了在加載數(shù)據(jù)的時(shí)候同時(shí)接收Router轉(zhuǎn)發(fā)來(lái)的Gateway Message，但是在數(shù)據(jù)加載完前這些受到的數(shù)據(jù)先被緩存起來(lái)，待所有 RoomGatewayList 數(shù)據(jù)加載完后就把這些數(shù)據(jù)重放一遍；

Broker之所以區(qū)分狀態(tài)，是為了在加載完畢 RoomGatewayList 數(shù)據(jù)前不對(duì)Proxy提供轉(zhuǎn)發(fā)消息的服務(wù)，同時(shí)也方便Broker Partition應(yīng)對(duì)的消息量增大時(shí)進(jìn)行水平擴(kuò)展。

當(dāng)Broker發(fā)生Partition擴(kuò)展的時(shí)候，新的Partition個(gè)數(shù)必須是2的冪，只有新Partition內(nèi)所有Broker Replica都加載實(shí)例完畢，再更改/pubsub/broker/partition_num的值。

老的Broker也要watch路徑/pubsub/broker/partition_num的值，當(dāng)這個(gè)值增加的時(shí)候，它也需要清洗本地的路由信息緩存。

Broker的擴(kuò)容過(guò)程猶如細(xì)胞分裂，形成中的兩個(gè)細(xì)胞有著完全相同的數(shù)據(jù)，分裂完成后【Registry路徑/pubsub/broker/partition_num的值翻倍】則需要清洗垃圾信息。這種方法稱為翻倍法。

4.7、Router

Router詳細(xì)流程如下：
?

• 1）Router加載配置，Registry地址；
• 2）把自身信息注冊(cè)到Registry路徑/pubsub/router下，把Registry返回的ReplicaID作為自身ID；
• 3）從Registry路徑/pubsub/broker/partition(x)下獲取每個(gè)Broker Partition的各個(gè)replica；
• 4）從Registry路徑/pubsub/broker/partition_num獲取當(dāng)前有效的Broker Partition Number；

?

• 5）啟動(dòng)一個(gè)線程關(guān)注Registry上的Broker路徑/pubsub/broker，以實(shí)時(shí)獲取以下信息：
• ? ? {Broker Partition Number}
• ? ? - 新的Broker Partition（此時(shí)發(fā)生了擴(kuò)容）；
• ? ? - Broker Partition內(nèi)新的broker replica（Partition內(nèi)發(fā)生了replica擴(kuò)容）；
• ? ? - Broker Parition內(nèi)某replica掛掉的信息；

?

• 6）定時(shí)向各個(gè)Broker Partition replica發(fā)送心跳，異步等待Broker返回的心跳響應(yīng)包，以探測(cè)其活性，以保證不向超時(shí)的replica轉(zhuǎn)發(fā)Gateway Message；
• 7）啟動(dòng)一個(gè)線程定時(shí)讀取Registry上的Broker路徑/pubsub/broker下各個(gè)子節(jié)點(diǎn)的值，以定時(shí)輪詢的策略觀察Broker Partition Number變動(dòng)，以及各Partition的變動(dòng)情況，作為實(shí)時(shí)策略的補(bǔ)充；同時(shí)定時(shí)檢查心跳包超時(shí)的Broker，從有效的BrokerList中刪除；
• 8）從Database全量加載路由 RoomGatewayList 數(shù)據(jù)放入本地緩存；
• 9）收取Gateway發(fā)來(lái)的心跳消息，及時(shí)返回ack包；
• 10）收取Gateway轉(zhuǎn)發(fā)來(lái)的Gateway Message，按照一定規(guī)則【BrokerPartitionID % BrokerPartitionNum = RoomID % BrokerPartitionNum】轉(zhuǎn)發(fā)給某個(gè)Broker Partition下所有Broker Replica，保證Partition下所有replica擁有同樣的路由 RoomGatewayList 數(shù)據(jù)，再把Message內(nèi)數(shù)據(jù)存入本地緩存，當(dāng)檢測(cè)到數(shù)據(jù)不重復(fù)的時(shí)候把數(shù)據(jù)異步寫入Database。

?

4.8、Gateway

Gateway詳細(xì)流程如下：
?

• 1）讀取配置文件，加載Registry地址；
• 2）從Registry路徑/pubsub/router/下獲取所有router replica，依據(jù)各Replica的ID遞增排序組成replica數(shù)組RouterArray；
• 3）啟動(dòng)一個(gè)線程實(shí)時(shí)關(guān)注Registry路徑/pubsub/router，以獲取Router的動(dòng)態(tài)變化，及時(shí)更新RouterArray；
• 4）啟動(dòng)一個(gè)線程定時(shí)輪詢獲取Registry路徑/pubsub/router下各個(gè)Router實(shí)例，作為關(guān)注策略的補(bǔ)充，以期本地RouterArray及時(shí)更新；定時(shí)給各個(gè)Router發(fā)送心跳，異步獲取心跳回包；定時(shí)清除RouterArray中心跳超時(shí)的Router成員；
• 5）當(dāng)有Room內(nèi)某成員客戶端連接上來(lái)或者Room內(nèi)所有成員都不連接當(dāng)前Gateway節(jié)點(diǎn)時(shí)，依據(jù)規(guī)則【RouterArrayIndex = RoomID % RouterNum】向某個(gè)Router發(fā)送Gateway Message；
• 6）收到Broker轉(zhuǎn)發(fā)來(lái)的Room Message時(shí)，根據(jù)MessageID進(jìn)行去重，如果不重復(fù)則把消息發(fā)送到連接到當(dāng)前Gateway的Room內(nèi)所有客戶端，同時(shí)把MessageID緩存起來(lái)以用于去重判斷。

Gateway本地有一個(gè)基于共享內(nèi)存的LRU Cache，存儲(chǔ)最近一段時(shí)間發(fā)送的消息的MessageID。

五、接下來(lái)迫切要解決的：系統(tǒng)穩(wěn)定性

系統(tǒng)具有了可擴(kuò)展性僅僅是系統(tǒng)可用的初步，整個(gè)系統(tǒng)要保證最低粒度的SLA（0.99），就必須在兩個(gè)維度對(duì)系統(tǒng)的可靠性就行感知：消息延遲和系統(tǒng)內(nèi)部組件的高可用。

5.1、消息延遲

準(zhǔn)確的消息延遲的統(tǒng)計(jì)，通用的做法可以基于日志系統(tǒng)對(duì)系統(tǒng)所有消息或者以一定概率抽樣后進(jìn)行統(tǒng)計(jì)，但限于人力目前沒(méi)有這樣做。

目前使用了一個(gè)方法：通過(guò)一種構(gòu)造一組偽用戶ID，定時(shí)地把消息發(fā)送給proxy，每條消息經(jīng)過(guò)一層就把在這層的進(jìn)入時(shí)間和發(fā)出時(shí)間以及組件自身的一些信息填入消息，這組偽用戶的消息最終會(huì)被發(fā)送到一個(gè)偽Gateway端，偽Gateway對(duì)這些消息的信息進(jìn)行歸并統(tǒng)計(jì)后，即可計(jì)算出當(dāng)前系統(tǒng)的平均消息延遲時(shí)間。

通過(guò)所有消息的平均延遲可以評(píng)估系統(tǒng)的整體性能。同時(shí)，因?yàn)橄到y(tǒng)消息路由的哈希方式已知，當(dāng)固定時(shí)間內(nèi)偽Gateway沒(méi)有收到消息時(shí)，就把消息當(dāng)做發(fā)送失敗，當(dāng)某條鏈路失敗一定次數(shù)后就可以產(chǎn)生告警了。

5.2、高可用

上面的方法同時(shí)能夠檢測(cè)某個(gè)鏈路是否出問(wèn)題，但是鏈路具體出問(wèn)題的點(diǎn)無(wú)法判斷，且實(shí)時(shí)性無(wú)法保證。

為了保證各個(gè)組件的高可用，系統(tǒng)引入了另一種評(píng)估方法：每個(gè)層次都給后端組件發(fā)送心跳包，通過(guò)心跳包的延遲和成功率判斷其下一級(jí)組件的當(dāng)前的可用狀態(tài)。

譬如proxy定時(shí)給每個(gè)Partition內(nèi)每個(gè)broker發(fā)送心跳，可以依據(jù)心跳的成功率來(lái)快速判斷broker是否處于“假死”狀態(tài)（最近業(yè)務(wù)就遇到過(guò)broker進(jìn)程還活著，但是對(duì)任何收到的消息都不處理的情況）。

同時(shí)依靠心跳包的延遲還可以判斷broker的處理能力，基于此延遲值可在同一Partition內(nèi)多broker端進(jìn)行負(fù)載均衡。

六、進(jìn)一步優(yōu)化：消息可靠性

公司內(nèi)部?jī)?nèi)部原有一個(gè)走tcp通道的群聊消息系統(tǒng)，但是經(jīng)過(guò)元旦一次大事故（幾乎全線崩潰）后，相關(guān)業(yè)務(wù)的一些重要消息改走這套基于UDP的群聊消息系統(tǒng)了。這些消息如服務(wù)端下達(dá)給客戶端的游戲動(dòng)作指令，是不允許丟失的，但其特點(diǎn)是相對(duì)于聊天消息來(lái)說(shuō)量非常小（單人1秒最多一個(gè)），所以需要在目前UDP鏈路傳遞消息的基礎(chǔ)之上再構(gòu)建一個(gè)可靠消息鏈路。

國(guó)內(nèi)某IM大廠的消息系統(tǒng)也是以UDP鏈路為基礎(chǔ)的（見(jiàn)《為什么QQ用的是UDP協(xié)議而不是TCP協(xié)議？》），他們的做法是消息重試加ack構(gòu)建了可靠消息穩(wěn)定傳輸鏈路。但是這種做法會(huì)降低系統(tǒng)的吞吐率，所以需要獨(dú)辟蹊徑。

UDP通信的本質(zhì)就是偽裝的IP通信，TCP自身的穩(wěn)定性無(wú)非是重傳、去重和ack，所以不考慮消息順序性的情況下可以通過(guò)重傳與去重來(lái)保證消息的可靠性。

基于目前系統(tǒng)的可靠消息傳輸流程如下：

• 1）Client給每個(gè)命令消息依據(jù)snowflake算法配置一個(gè)ID，復(fù)制三份，立即發(fā)送給不同的Proxy；
• 2）Proxy收到命令消息以后隨機(jī)發(fā)送給一個(gè)Broker；
• 3）Broker收到后傳輸給Gateway；
• 4）Gateway接收到命令消息后根據(jù)消息ID進(jìn)行重復(fù)判斷，如果重復(fù)則丟棄，否則就發(fā)送給APP，并緩存之。

正常的消息在群聊消息系統(tǒng)中傳輸時(shí)，Proxy會(huì)根據(jù)消息的Room ID傳遞給固定的Broker，以保證消息的有序性。

七、Router需要進(jìn)一步強(qiáng)化

?

7.1、簡(jiǎn)述

當(dāng)線上需要部署多套群聊消息系統(tǒng)的時(shí)候，Gateway需要把同樣的Room Message復(fù)制多份轉(zhuǎn)發(fā)給多套群聊消息系統(tǒng)，會(huì)增大Gateway壓力，可以把Router單獨(dú)獨(dú)立部署，然后把Room Message向所有的群聊消息系統(tǒng)轉(zhuǎn)發(fā)。

Router系統(tǒng)原有流程是：Gateway按照Room ID把消息轉(zhuǎn)發(fā)給某個(gè)Router，然后Router把消息轉(zhuǎn)發(fā)給下游Broker實(shí)例。新部署一套群聊消息系統(tǒng)的時(shí)候，新系統(tǒng)Broker的schema需要通過(guò)一套約定機(jī)制通知Router，使得Router自身邏輯過(guò)于復(fù)雜。

重構(gòu)后的Router架構(gòu)參照上圖，也采用分Partition分Replica設(shè)計(jì)，Partition內(nèi)部各Replica之間采用non-leader機(jī)制；各Router Replica不會(huì)主動(dòng)把Gateway Message內(nèi)容push給各Broker，而是各Broker主動(dòng)通過(guò)心跳包形式向Router Partition內(nèi)某個(gè)Replica注冊(cè)，而后此Replica才會(huì)把消息轉(zhuǎn)發(fā)到這個(gè)Broker上。

類似于Broker，Router Partition也以2倍擴(kuò)容方式進(jìn)行Partition水平擴(kuò)展，并通過(guò)一定機(jī)制保證擴(kuò)容或者Partition內(nèi)部各個(gè)實(shí)例停止運(yùn)行或者新啟動(dòng)時(shí)，盡力保證數(shù)據(jù)的一致性。

Router Replica收到Gateway Message后，replica先把Gateway Message轉(zhuǎn)發(fā)給Partition內(nèi)各個(gè)peer replica，然后再轉(zhuǎn)發(fā)給各個(gè)訂閱者。Router轉(zhuǎn)發(fā)消息的同時(shí)異步把消息數(shù)據(jù)寫入Database。

獨(dú)立Router架構(gòu)下，下面小節(jié)將分別詳述Gateway、Router和Broker三個(gè)相關(guān)模塊的詳細(xì)流程。

7.2、Gateway

Gateway詳細(xì)流程如下：
?

• 1）從Registry路徑/pubsub/router/partition(x)下獲取每個(gè)Partition的各個(gè)replica；
• 2）從Registry路徑/pubsub/router/partition_num獲取當(dāng)前有效的Router Partition Number；
• 3）啟動(dòng)一個(gè)線程關(guān)注Registry上的Router路徑/pubsub/router，以實(shí)時(shí)獲取以下信息：{Router Partition Number} -> 新的Router Partition（此時(shí)發(fā)生了擴(kuò)容）；??Partition內(nèi)新的replica（Partition內(nèi)發(fā)生了replica擴(kuò)容）；??Parition內(nèi)某replica掛掉的信息；
• 4）定時(shí)向各個(gè)Partition replica發(fā)送心跳，異步等待Router返回的心跳響應(yīng)包，以探測(cè)其活性，以保證不向超時(shí)的replica轉(zhuǎn)發(fā)Gateway Message；
• 5）啟動(dòng)一個(gè)線程定時(shí)讀取Registry上的Router路徑/pubsub/router下各個(gè)子節(jié)點(diǎn)的值，以定時(shí)輪詢的策略觀察Router Partition Number變動(dòng)，以及各Partition的變動(dòng)情況，作為實(shí)時(shí)策略的補(bǔ)充；同時(shí)定時(shí)檢查心跳包超時(shí)的Router，從有效的BrokerList中刪除；
• 6 依據(jù)規(guī)則向某個(gè)Partition的replica轉(zhuǎn)發(fā)Gateway Message。

第六步的規(guī)則決定了Gateway Message的目的Partition和replica，規(guī)則內(nèi)容有：

如果某Router Partition ID滿足condition(RoomID % RouterPartitionNumber == RouterPartitionID % RouterPartitionNumber)，則把消息轉(zhuǎn)發(fā)到此Partition；

這里之所以不采用直接hash方式(RouterPartitionID = RoomID % RouterPartitionNumber)獲取Router Partition，是考慮到當(dāng)Router進(jìn)行2倍擴(kuò)容的時(shí)候當(dāng)所有新的Partition的所有Replica都啟動(dòng)完畢且數(shù)據(jù)一致時(shí)才會(huì)修改Registry路徑/pubsub/router/partitionnum的值，按照規(guī)則的計(jì)算公式才能保證新Partition的各個(gè)Replica在啟動(dòng)過(guò)程中就可以得到Gateway Message，也即此時(shí)每個(gè)Gateway Message會(huì)被發(fā)送到兩個(gè)Router Partition。 當(dāng)Router擴(kuò)容完畢，修改Registry路徑/pubsub/router/partitionnum的值后，此時(shí)新集群進(jìn)入穩(wěn)定期，每個(gè)Gateway Message只會(huì)被發(fā)送固定的一個(gè)Partition，condition(RoomID % RouterPartitionNumber == RouterPartitionID % RouterPartitionNumber)等效于condition(RouterPartitionID = RoomID % RouterPartitionNumber)。

如果Router Partition內(nèi)某replia滿足condition(replicaPartitionID = RoomID % RouterPartitionReplicaNumber)，則把消息轉(zhuǎn)發(fā)到此replica。

replica向Registry注冊(cè)的時(shí)候得到的ID稱之為replicaID，Router Parition內(nèi)所有replica按照replicaID遞增排序組成replica數(shù)組RouterPartitionReplicaArray，replicaPartitionID即為replica在數(shù)組中的下標(biāo)。

Gateway Message數(shù)據(jù)一致性：

Gateway向Router發(fā)送的Router Message內(nèi)容有兩種：某user在當(dāng)前Gateway上進(jìn)入某Room 和 某user在當(dāng)前Gateway上退出某Room，數(shù)據(jù)項(xiàng)分別是UIN（用戶ID）、Room ID、Gateway Addr和User Action(Login or Logout。

由于所有消息都是走UDP鏈路進(jìn)行轉(zhuǎn)發(fā)，則這些消息的順序就有可能亂序。Gateway可以統(tǒng)一給其發(fā)出的所有消息分配一個(gè)全局遞增的ID【下文稱為GatewayMsgID，Gateway Message ID】以保證消息的唯一性和全局有序性。

Gateway向Registry注冊(cè)臨時(shí)有序節(jié)點(diǎn)時(shí)，Registry會(huì)給Gateway分配一個(gè)ID，Gateway可以用這個(gè)ID作為自身的Instance ID【假設(shè)這個(gè)ID上限是65535】。

GatewayMsgID字長(zhǎng)是64bit，其格式如下：

1 2	// 63 -------------------------- 48 47 -------------- 38 37 ------------ 0 // |? 16bit Gateway Instance ID??? |?? 10bit Reserve??? |??? 38bit自增碼? |

?

7.3、Router

Router系統(tǒng)部署之前，先設(shè)置Registry路徑/pubsub/router/partition_num的值為1。

Router詳細(xì)流程如下：
?

• 1）Router加載配置，獲取自身所在Partition的ID（假設(shè)為3）；
• 2）向Registry路徑/pubsub/router/partition3注冊(cè)，設(shè)置其狀態(tài)為Init，注冊(cè)中心返回的ID作為自身的ID(replicaID)；
• 3）注冊(cè)完畢會(huì)收到Gateway發(fā)來(lái)的Gateway Message以及Broker發(fā)來(lái)的心跳消息（HeartBeat Message），先緩存到消息隊(duì)列MessageQueue；
• 4）從Registry路徑/pubsub/router/partition3下獲取自身所在的Partition內(nèi)的各個(gè)replica；
• 5）從Registry路徑/pubsub/router/partition_num獲取當(dāng)前有效的Router Partition Number；
• 6）啟動(dòng)一個(gè)線程關(guān)注Registry路徑/pubsub/router，以實(shí)時(shí)獲取以下信息：{Router Partition Number}??-> Partition內(nèi)新的replica（Partition內(nèi)發(fā)生了replica擴(kuò)容）；??Parition內(nèi)某replica掛掉的信息；
• 7）從Database加載數(shù)據(jù)；
• 8）啟動(dòng)一個(gè)線程異步處理MessageQueue內(nèi)的Gateway Message，把Gateway Message轉(zhuǎn)發(fā)給同Partition內(nèi)其他peer replica，然后依據(jù)規(guī)則【RoomID % BrokerPartitionNumber == BrokerReplicaPartitionID % BrokerPartitionNumber】轉(zhuǎn)發(fā)給BrokerList內(nèi)每個(gè)Broker；處理Broker發(fā)來(lái)的心跳包，把Broker的信息存入本地BrokerList，然后給Broker發(fā)送回包；
• 9）修改Registry路徑/pubsub/router/partition3下節(jié)點(diǎn)的狀態(tài)為Running；
• 10）啟動(dòng)一個(gè)線程定時(shí)讀取Registry路徑/pubsub/router下各個(gè)子路徑的值，以定時(shí)輪詢的策略觀察Router各Partition的變動(dòng)情況，作為實(shí)時(shí)策略的補(bǔ)充；檢查超時(shí)的Broker，把其從BrokerList中剔除；
• 11）當(dāng)RouterPartitionNum倍增時(shí)，Router依據(jù)規(guī)則【RoomID % BrokerPartitionNumber == BrokerReplicaPartitionID % BrokerPartitionNumber】清洗自身路由信息緩存中數(shù)據(jù)；
• 12）Router本地存儲(chǔ)每個(gè)Gateway的最大GatewayMsgID，收到小于GatewayMsgID的Gateway Message可以丟棄不處理，否則就更新GatewayMsgID并根據(jù)上面邏輯進(jìn)行處理。

之所以把Gateway Message和Heartbeat Message放在一個(gè)線程處理，是為了防止進(jìn)程假死這種情況。

Broker也采用了分Partition分Replica機(jī)制，所以向Broker轉(zhuǎn)發(fā)Gateway Message時(shí)候路由規(guī)則，與Gateway向Router轉(zhuǎn)發(fā)消息的路由規(guī)則相同。

另外啟動(dòng)一個(gè)工具，當(dāng)水平擴(kuò)展后新啟動(dòng)的Partition內(nèi)所有Replica的狀態(tài)都是Running的時(shí)候，修改Registry路徑/pubsub/router/partition_num的值為所有Partition的數(shù)目。

7.4、Broker

Broker詳細(xì)流程如下：
?

• 1）Broker加載配置，獲取自身所在Partition的ID（假設(shè)為3）；
• 2）向Registry路徑/pubsub/broker/partition3注冊(cè)，設(shè)置其狀態(tài)為Init，注冊(cè)中心返回的ID作為自身的ID(replicaID)；
• 3）從Registry路徑/pubsub/router/partition_num獲取當(dāng)前有效的Router Partition Number；
• 4）從Registry路徑/pubsub/router/partition(x)下獲取每個(gè)Router Partition的各個(gè)replica；
• 5）啟動(dòng)一個(gè)線程關(guān)注Registry路徑/pubsub/router，以實(shí)時(shí)獲取以下信息：{Router Partition Number} -> 新的Router Partition（此時(shí)發(fā)生了擴(kuò)容）；??Partition內(nèi)新的replica（Partition內(nèi)發(fā)生了replica擴(kuò)容）；??Parition內(nèi)某replica掛掉的信息；
• 6）依據(jù)規(guī)則【RouterPartitionID % BrokerPartitionNum == BrokerPartitionID % BrokerPartitionNum，RouterReplicaID = BrokerReplicaID % BrokerPartitionNum】選定目標(biāo)Router Partition下某個(gè)Router replica，向其發(fā)送心跳消息，包含BrokerPartitionNum、BrokerPartitionID、BrokerHostAddr和精確到秒級(jí)的Timestamp，并異步等待所有Router replica的回復(fù)，所有Router轉(zhuǎn)發(fā)來(lái)的Gateway Message放入GatewayMessageQueue；
• 7）依據(jù)規(guī)則【BrokerPartitionID == RoomID % BrokerParitionNum】從Database加載數(shù)據(jù)；
• 8）依據(jù)規(guī)則【BrokerPartitionID % BrokerParitionNum == RoomID % BrokerParitionNum】異步處理GatewayMessageQueue內(nèi)的Gateway Message，只留下合乎規(guī)則的消息的數(shù)據(jù)；
• 9）修改Registry路徑/pubsub/broker/partition3下自身節(jié)點(diǎn)的狀態(tài)為Running；
• 10）啟動(dòng)一個(gè)線程定時(shí)讀取Registry路徑/pubsub/router下各個(gè)子路徑的值，以定時(shí)輪詢的策略觀察Router各Partition的變動(dòng)情況，作為實(shí)時(shí)策略的補(bǔ)充；定時(shí)檢查超時(shí)的Router，某Router超時(shí)后更換其所在的Partition內(nèi)其他Router替換之，定時(shí)發(fā)送心跳包；
• 11）當(dāng)Registry路徑/pubsub/broker/partition_num的值BrokerPartitionNum發(fā)生改變的時(shí)候，依據(jù)規(guī)則【PartitionID == RoomID % PartitionNum】清洗本地路由信息緩存中每條數(shù)據(jù)；
• 12）接收Proxy發(fā)來(lái)的Room Message，依據(jù)RoomID從路由信息緩存中查找Room有成員登陸的所有Gateway，把消息轉(zhuǎn)發(fā)給這些Gateway；
• 13）Broker本地存儲(chǔ)每個(gè)Gateway的最大GatewayMsgID，收到小于GatewayMsgID的Gateway Message可以丟棄不處理，否則更新GatewayMsgID并根據(jù)上面邏輯進(jìn)行處理。

BrokerPartitionNumber可以小于或者等于或者大于RouterPartitionNumber，兩個(gè)數(shù)應(yīng)該均是2的冪，兩個(gè)集群可以分別進(jìn)行擴(kuò)展，互不影響。譬如BrokerPartitionNumber=4而RouterPartitionNumber=2，則Broker Partition 3只需要向Router Partition 1的某個(gè)follower發(fā)送心跳消息即可；若BrokerPartitionNumber=4而RouterPartitionNumber=8，則Broker Partition 3需要向Router Partition 3的某個(gè)follower發(fā)送心跳消息的同時(shí)，還需要向Router Partition 7的某個(gè)follower發(fā)送心跳，以獲取全量的Gateway Message。

Broker需要關(guān)注/pubsub/router/partitionnum和/pubsub/broker/partitionnum的值的變化，當(dāng)router或者broker進(jìn)行parition水平擴(kuò)展的時(shí)候，Broker需要及時(shí)重新構(gòu)建與Router之間的對(duì)應(yīng)關(guān)系，及時(shí)變動(dòng)發(fā)送心跳的Router Replica對(duì)象【RouterPartitionID = BrokerReplicaID % RouterPartitionNum，RouterPartitionID為Router Replica在PartitionRouterReplicaArray數(shù)組的下標(biāo)】。

當(dāng)Router Partition內(nèi)replica死掉或者發(fā)送心跳包的replica對(duì)象死掉（無(wú)論是注冊(cè)中心通知還是心跳包超時(shí)），broker要及時(shí)變動(dòng)發(fā)送心跳的Router replica對(duì)象。

另外，Gateway使用UDP通信方式向Router發(fā)送Gateway Message，如若這個(gè)Message丟失則此Gateway上該Room內(nèi)所有成員一段時(shí)間內(nèi)（當(dāng)有新的成員在當(dāng)前Gateway上加入Room 時(shí)會(huì)產(chǎn)生新的Gateway Message）都無(wú)法再接收消息，為了保證消息的可靠性，可以使用這樣一個(gè)約束解決問(wèn)題：在此Gateway上登錄的某Room內(nèi)的人數(shù)少于3時(shí)，Gateway會(huì)把Gateway Message復(fù)制兩份非連續(xù)（如以10ms為時(shí)間間隔）重復(fù)發(fā)送給某個(gè)Partition leader。因Gateway Message消息處理的冪等性，重復(fù)Gateway Message并不會(huì)導(dǎo)致Room Message發(fā)送錯(cuò)誤，只在極少概率的情況下會(huì)導(dǎo)致Gateway收到消息的時(shí)候Room內(nèi)已經(jīng)沒(méi)有成員在此Gateway登錄，此時(shí)Gateway會(huì)把消息丟棄不作處理。

傳遞實(shí)時(shí)消息群聊消息系統(tǒng)的Broker向特定Gateway轉(zhuǎn)發(fā)Room Message的時(shí)候，會(huì)帶上Room內(nèi)在此Gateway上登錄的用戶列表，Gateway根據(jù)這個(gè)用戶列表下發(fā)消息時(shí)如果檢測(cè)到此用戶已經(jīng)下線，在放棄向此用戶轉(zhuǎn)發(fā)消息的同時(shí)，還應(yīng)該把此用戶已經(jīng)下線的消息發(fā)送給Router，當(dāng)Router把這個(gè)消息轉(zhuǎn)發(fā)給Broker后，Broker把此用戶從用戶列表中剔除。通過(guò)這種負(fù)反饋機(jī)制保證用戶狀態(tài)更新的及時(shí)性。

八、離線消息的處理

?

8.1、簡(jiǎn)述

前期的系統(tǒng)只考慮了用戶在線情況下實(shí)時(shí)消息的傳遞，當(dāng)用戶離線時(shí)其消息便無(wú)法獲取。

若系統(tǒng)考慮用戶離線消息傳遞，需要考慮如下因素：
?

• 1）消息固化：保證用戶上線時(shí)收到其離線期間的消息；
• 2）消息有序：離線消息和在線消息都在一個(gè)消息系統(tǒng)傳遞，給每個(gè)消息分配一個(gè)ID以區(qū)分消息先后順序，消息順序越靠后則ID愈大。

離線消息的存儲(chǔ)和傳輸，需要考慮用戶的狀態(tài)以及每條消息的發(fā)送狀態(tài)，整個(gè)消息核心鏈路流程會(huì)有大的重構(gòu)。

新消息架構(gòu)如下圖：

?

系統(tǒng)名詞解釋：
?

• 1）Pi : 消息ID存儲(chǔ)模塊，存儲(chǔ)每個(gè)人未發(fā)送的消息ID有序遞增集合；
• 2）Xiu : 消息存儲(chǔ)KV模塊，存儲(chǔ)每個(gè)人的消息，給每個(gè)消息分配ID，以ID為key，以消息內(nèi)為value；
• 3）Gateway Message(HB) : 用戶登錄登出消息，包括APP保活定時(shí)心跳（Hearbeat）消息。

系統(tǒng)內(nèi)部代號(hào)貔貅(貔貅者，雄貔雌貅)，源自上面兩個(gè)新模塊。

這個(gè)版本架構(gòu)流程的核心思想為“消息ID與消息內(nèi)容分離，消息與用戶狀態(tài)分離”。消息發(fā)送流程涉及到模塊 Client/Proxy/Pi/Xiu，消息推送流程則涉及到模塊 Pi/Xiu/Broker/Router/Gateway。

下面小節(jié)先細(xì)述Pi和Xiu的接口，然后再詳述發(fā)送和推送流程。

8.2、Xiu

Xiu模塊功能名稱是Message Storage，用戶緩存和固化消息，并給消息分配ID。Xiu 集群采用分 Partition 分 Replica 機(jī)制，Partition 初始數(shù)目須是2的倍數(shù)，集群擴(kuò)容時(shí)采用翻倍法。
?

8.2.1存儲(chǔ)消息

存儲(chǔ)消息請(qǐng)求的參數(shù)列表為{SnowflakeID，UIN, Message}，其流程如下：
?

• 1）接收客戶端發(fā)來(lái)的消息，獲取消息接收人ID（UIN）和客戶端給消息分配的 SnowflakeID；
• 2）檢查 UIN % Xiu_Partition_Num == Xiu_Partition_ID % Xiu_Partition_Num 添加是否成立【即接收人的消息是否應(yīng)當(dāng)由當(dāng)前Xiu負(fù)責(zé)】，不成立則返回錯(cuò)誤并退出；
• 3）檢查 SnowflakeID 對(duì)應(yīng)的消息是否已經(jīng)被存儲(chǔ)過(guò)，若已經(jīng)存儲(chǔ)過(guò)則返回其對(duì)應(yīng)的消息ID然后退出；
• 4）給消息分配一個(gè) MsgID：
  每個(gè)Xiu有自己唯一的 Xiu_Partition_ID，以及一個(gè)初始值為 0 的 Partition_Msg_ID。MsgID = 1B[ Xiu_Partition_ID ] + 1B[ Message Type ] + 6B[ ++ Partition_Msg_ID ]。每次分配的時(shí)候 Partition_Msg_ID 都自增加一。
• 5）以 MsgID 為 key 把消息存入基于共享內(nèi)存的 Hashtable，并存入消息的 CRC32 hash值和插入時(shí)間，把 MsgID 存入一個(gè) LRU list 中：
  LRU List 自身并不存入共享內(nèi)存中，當(dāng)進(jìn)程重啟時(shí)，可以根據(jù)Hashtable中的數(shù)據(jù)重構(gòu)出這個(gè)List。把消息存入 Hashtable 中時(shí)，如果 Hashtable full，則依據(jù) LRU List 對(duì)Hashtable 中的消息進(jìn)行淘汰。
• 6）把MsgID返回給客戶端；
• 7）把MsgID異步通知給消息固化線程，消息固化線程根據(jù)MsgID從Hashtable中讀取消息并根據(jù)CRC32 hash值判斷消息內(nèi)容是否完整，完整則把消息存入本地RocksDB中。

?

8.2.2讀取消息

讀取消息請(qǐng)求的參數(shù)列表為{UIN, MsgIDList}，其流程為：
?

• 1）獲取請(qǐng)求的 MsgIDList，判斷每個(gè)MsgID MsgID{Xiu_Partition_ID} == Xiu_Partition_ID 條件是否成立，不成立則返回錯(cuò)誤并退出；
• 2）從 Hashtable 中獲取每個(gè) MsgID 對(duì)應(yīng)的消息；
• 3）如果 Hashtable 中不存在，則從 RocksDB 中讀取 MsgID 對(duì)應(yīng)的消息；
• 4）讀取完畢則把所有獲取的消息返回給客戶端。

?

8.2.3主從數(shù)據(jù)同步

目前從簡(jiǎn)，暫定Xiu的副本只有一個(gè)。

Xiu節(jié)點(diǎn)啟動(dòng)的時(shí)候根據(jù)自身配置文件中分配的 Xiu_Partition_ID 到Registry路徑 /pubsub/xiu/partition_id 下進(jìn)行注冊(cè)一個(gè)臨時(shí)有序節(jié)點(diǎn)，注冊(cè)成功則Registry會(huì)返回Xiu的節(jié)點(diǎn) ID。

Xiu節(jié)點(diǎn)獲取 /pubsub/xiu/partition_id 下的所有節(jié)點(diǎn)的ID和地址信息，依據(jù) 節(jié)點(diǎn)ID最小者為leader 的原則，即可判定自己的角色。只有l(wèi)eader可接受讀寫數(shù)據(jù)請(qǐng)求。

數(shù)據(jù)同步流程如下：
?

• 1）follower定時(shí)向leader發(fā)送心跳信息，心跳信息包含本地最新消息的ID；
• 2）leader啟動(dòng)一個(gè)數(shù)據(jù)同步線程處理follower的心跳信息，leader的數(shù)據(jù)同步線程從LRU list中查找 follower_latest_msg_id 之后的N條消息的ID，若獲取到則讀取消息并同步給follower，獲取不到則回復(fù)其與leader之間消息差距太大；
• 3）follower從leader獲取到最新一批消息，則存儲(chǔ)之；
• 4）follower若獲取leader的消息差距太大響應(yīng)，則請(qǐng)求leader的agent把RocksDB的固化數(shù)據(jù)全量同步過(guò)來(lái)，整理完畢后再次啟動(dòng)與leader之間的數(shù)據(jù)同步流程。

follower會(huì)關(guān)注Registry路徑 /pubsub/xiu/partition_id 下所有所有節(jié)點(diǎn)的變化情況，如果leader掛掉則及時(shí)轉(zhuǎn)換身份并接受客戶端請(qǐng)求。如果follower 與 leader 之間的心跳超時(shí)，則follower刪掉 leader 的 Registry 路徑節(jié)點(diǎn)，及時(shí)進(jìn)行身份轉(zhuǎn)換處理客戶端請(qǐng)求。

當(dāng)leader重啟或者follower轉(zhuǎn)換為leader的時(shí)候，需要把 Partition_Msg_ID 進(jìn)行一個(gè)大數(shù)值增值（譬如增加1000）以防止可能的消息ID亂序情況。
?

8.2.4集群擴(kuò)容

Xiu 集群擴(kuò)容采用翻倍法，擴(kuò)容時(shí)新 Partition 的節(jié)點(diǎn)啟動(dòng)后工作流程如下：
?

• 1）向Registry的路徑 /pubsub/xiu/partition_id 下自己的 node 的 state 為 running，同時(shí)注冊(cè)自己的對(duì)外服務(wù)地址信息；
• 2）另外啟動(dòng)一個(gè)工具，當(dāng)水平擴(kuò)展后所有新啟動(dòng)的 Partition 內(nèi)所有 Replica 的狀態(tài)都是 Running 的時(shí)候，修改 Registry 路徑 /pubsub/xiu/partition_num 的值為擴(kuò)容后 Partition 的數(shù)目。按照開頭的例子，即由2升級(jí)為4。

之所以 Xiu 不用像 Broker 和 Router 那樣啟動(dòng)的時(shí)候向老的 Partition 同步數(shù)據(jù)，是因?yàn)槊總€(gè) Xiu 分配的 MsgID 中已經(jīng)帶有 Xiu 的 PartitionID 信息，即使集群擴(kuò)容這個(gè) ID 也不變，根據(jù)這個(gè)ID也可以定位到其所在的Partition，而不是借助 hash 方法。

8.3、Pi

Pi 模塊功能名稱是 Message ID Storage，存儲(chǔ)每個(gè)用戶的 MsgID List。Xiu 集群也采用分 Partition 分 Replica 機(jī)制，Partition 初始數(shù)目須是2的倍數(shù)，集群擴(kuò)容時(shí)采用翻倍法。
?

8.3.1存儲(chǔ)消息ID

MsgID 存儲(chǔ)的請(qǐng)求參數(shù)列表為{UIN，MsgID}，Pi 工作流程如下：
?

• 1）判斷條件 UIN % Pi_Partition_Num == Pi_Partition_ID % Pi_Partition_Num 是否成立，若不成立則返回error退出；
• 2）把 MsgID 插入U(xiǎn)IN的 MsgIDList 中，保持 MsgIDList 中所有 MsgID 不重復(fù)有序遞增，把請(qǐng)求內(nèi)容寫入本地log，給請(qǐng)求者返回成功響應(yīng)。

Pi有專門的日志記錄線程，給每個(gè)日志操作分配一個(gè) LogID，每個(gè) Log 文件記錄一定量的寫操作，當(dāng)文件 size 超過(guò)配置的上限后刪除之。
?

8.3.2讀取消息ID列表

讀取請(qǐng)求參數(shù)列表為{UIN, StartMsgID, MsgIDNum, ExpireFlag}，其意義為獲取用戶 UIN 自起始ID為 StartMsgID 起（不包括 StartMsgID ）的數(shù)目為 MsgIDNum 的消息ID列表，ExpireFlag意思是 所有小于等于 StartMsgID 的消息ID是否刪除。?

流程如下：
?

• 1）判斷條件 UIN % Pi_Partition_Num == Pi_Partition_ID % Pi_Partition_Num 是否成立，若不成立則返回error退出；
• 2）獲取 (StartID, StartMsgID + MsgIDNum] 范圍內(nèi)的所有 MsgID，把結(jié)果返回給客戶端；
• 3）如果 ExpireFlag 有效，則刪除MsgIDList內(nèi)所有在 [0, StartMsgID] 范圍內(nèi)的MsgID，把請(qǐng)求內(nèi)容寫入本地log。

?

8.3.3主從數(shù)據(jù)同步

同 Xiu 模塊，暫定 Pi 的同 Parition 副本只有一個(gè)。

Pi 節(jié)點(diǎn)啟動(dòng)的時(shí)候根據(jù)自身配置文件中分配的 Pi_Partition_ID 到Registry路徑 /pubsub/pi/partition_id 下進(jìn)行注冊(cè)一個(gè)臨時(shí)有序節(jié)點(diǎn)，注冊(cè)成功則 Registry 會(huì)返回 Pi 的節(jié)點(diǎn) ID。

Pi 節(jié)點(diǎn)獲取 /pubsub/pi/partition_id 下的所有節(jié)點(diǎn)的ID和地址信息，依據(jù) 節(jié)點(diǎn)ID最小者為leader 的原則，即可判定自己的角色。只有 leader 可接受讀寫數(shù)據(jù)請(qǐng)求。

數(shù)據(jù)同步流程如下：
?

• 1）follower 定時(shí)向 leader 發(fā)送心跳信息，心跳信息包含本地最新 LogID；
• 2）leader 啟動(dòng)一個(gè)數(shù)據(jù)同步線程處理 follower 的心跳信息，根據(jù) follower 匯報(bào)的 logID 把此 LogID；
• 3）follower 從 leader 獲取到最新一批 Log，先存儲(chǔ)然后重放。

follower 會(huì)關(guān)注Registry路徑 /pubsub/pi/partition_id 下所有節(jié)點(diǎn)的變化情況，如果 leader 掛掉則及時(shí)轉(zhuǎn)換身份并接受客戶端請(qǐng)求。如果follower 與 leader 之間的心跳超時(shí)，則follower刪掉 leader 的 Registry 路徑節(jié)點(diǎn)，及時(shí)進(jìn)行身份轉(zhuǎn)換處理客戶端請(qǐng)求。
?

8.3.4集群擴(kuò)容

Pi 集群擴(kuò)容采用翻倍法。則節(jié)點(diǎn)啟動(dòng)后工作流程如下：
?

• 1）向 Registry 注冊(cè)，獲取 Registry 路徑 /pubsub/xiu/partition_num 的值 PartitionNumber；
• 2）如果發(fā)現(xiàn)自己 PartitionID 滿足條件 PartitionID >= PartitionNumber 時(shí)，則意味著當(dāng)前 Partition 是擴(kuò)容后的新集群，更新 Registry 中自己狀態(tài)為start；
• 3）讀取 Registry 路徑 /pubsub/xiu 下所有 Parition 的 leader，根據(jù)條件 自身PartitionID % PartitionNumber == PartitionID % PartitionNumber 尋找對(duì)應(yīng)的老 Partition 的 leader，稱之為 parent_leader；
• 4）緩存收到 Proxy 轉(zhuǎn)發(fā)來(lái)的用戶請(qǐng)求；
• 5）向 parent_leader 獲取log；
• 6）向 parent_leader 同步內(nèi)存數(shù)據(jù)；
• 7）重放 parent_leader 的log；
• 8）更新 Registry 中自己的狀態(tài)為 Running；
• 9）重放用戶請(qǐng)求；
• 10）當(dāng) Registry 路徑 /pubsub/xiu/partition_num 的值 PartitionNumber 滿足條件 PartitionID >= PartitionNumber 時(shí)，意味著擴(kuò)容完成，處理用戶請(qǐng)求時(shí)要給用戶返回響應(yīng)。

Proxy 會(huì)把讀寫請(qǐng)求參照條件 UIN % Pi\_Partition\_Num == Pi\_Partition\_ID % Pi\_Partition\_Num 向相關(guān) partition 的 leader 轉(zhuǎn)發(fā)用戶請(qǐng)求。假設(shè)原來(lái) PartitionNumber 值為2，擴(kuò)容后值為4，則原來(lái)轉(zhuǎn)發(fā)給 partition0 的寫請(qǐng)求現(xiàn)在需同時(shí)轉(zhuǎn)發(fā)給 partition0 和 partition2，原來(lái)轉(zhuǎn)發(fā)給 partition1 的寫請(qǐng)求現(xiàn)在需同時(shí)轉(zhuǎn)發(fā)給 partition1 和 partition3。

另外啟動(dòng)一個(gè)工具，當(dāng)水平擴(kuò)展后所有新啟動(dòng)的 Partition 內(nèi)所有 Replica 的狀態(tài)都是 Running 的時(shí)候，修改Registry路徑/pubsub/xiu/partition_num的值為擴(kuò)容后 Partition 的數(shù)目。

8.4、數(shù)據(jù)發(fā)送流程

消息自 PiXiu 的外部客戶端（Client，服務(wù)端所有使用 PiXiu 提供的服務(wù)者統(tǒng)稱為客戶端）按照一定負(fù)載均衡規(guī)則發(fā)送到 Proxy，然后存入 Xiu 中，把 MsgID 存入 Pi 中。

其詳細(xì)流程如下：
?

• 1）Client 依據(jù) snowflake 算法給消息分配 SnowflakeID，依據(jù) ProxyID = UIN % ProxyNum 規(guī)則把消息發(fā)往某個(gè) Proxy；
• 2）Proxy 收到消息后轉(zhuǎn)發(fā)到 Xiu；
• 3）Proxy 收到 Xiu 返回的響應(yīng)后，把響應(yīng)轉(zhuǎn)發(fā)給 Client；
• 4）如果 Proxy 收到 Xiu 返回的響應(yīng)帶有 MsgID，則發(fā)起 Pi 寫流程，把 MsgID 同步到 Pi 中；
• 5）如果 Proxy 收到 Xiu 返回的響應(yīng)帶有 MsgID，則給 Broker 發(fā)送一個(gè) Notify，告知其某 UIN 的最新 MsgID。

?

8.5、數(shù)據(jù)轉(zhuǎn)發(fā)流程

轉(zhuǎn)發(fā)消息的主體是Broker，原來(lái)的在線消息轉(zhuǎn)發(fā)流程是它收到 Proxy 轉(zhuǎn)發(fā)來(lái)的 Message，然后根據(jù)用戶是否在線然后轉(zhuǎn)發(fā)給 Gateway。

PiXiu架構(gòu)下 Broker 會(huì)收到以下類型消息：
?

• 1）用戶登錄消息；
• 2）用戶心跳消息；
• 3）用戶登出消息；
• 4）Notify 消息；
• 5）Ack 消息。

Broker流程受這五種消息驅(qū)動(dòng)，下面分別詳述其收到這五種消息時(shí)的處理流程。

用戶登錄消息流程如下：
?

• 1）檢查用戶的當(dāng)前狀態(tài)，若為 OffLine 則把其狀態(tài)值為在線 OnLine；
• 2）檢查用戶的待發(fā)送消息隊(duì)列是否為空，不為空則退出；
• 3）向 Pi 模塊發(fā)送獲取 N 條消息 ID 的請(qǐng)求 {UIN: uin, StartMsgID: 0, MsgIDNum: N, ExpireFlag: false}，設(shè)置用戶狀態(tài)為 GettingMsgIDList 并等待回應(yīng)；
• 4）根據(jù) Pi 返回的消息 ID 隊(duì)列，向 Xiu 發(fā)起獲取消息請(qǐng)求 {UIN: uin, MsgIDList: msg ID List}，設(shè)置用戶狀態(tài)為 GettingMsgList 并等待回應(yīng)；
• 5）Xiu 返回消息列表后，設(shè)置狀態(tài)為 SendingMsg，并向 Gateway 轉(zhuǎn)發(fā)消息。

可以把用戶心跳消息當(dāng)做用戶登錄消息處理。

Gateway的用戶登出消息產(chǎn)生有三種情況：
?

• 1）用戶主動(dòng)退出；
• 2）用戶心跳超時(shí)；
• 3）給用戶轉(zhuǎn)發(fā)消息時(shí)發(fā)生網(wǎng)絡(luò)錯(cuò)誤。

用戶登出消息處理流程如下：
?

• 1）檢查用戶狀態(tài)，如果為 OffLine，則退出；
• 2）用戶狀態(tài)不為 OffLine 且檢查用戶已經(jīng)發(fā)送出去的消息列表的最后一條消息的 ID（LastMsgID），向 Pi 發(fā)送獲取 MsgID 請(qǐng)求{UIN: uin, StartMsgID: LastMsgID, MsgIDNum: 0, ExpireFlag: True}，待 Pi 返回響應(yīng)后退出。

處理 Proxy 發(fā)來(lái)的 Notify 消息處理流程如下：
?

• 1）如果用戶狀態(tài)為 OffLine，則退出；
• 2）更新用戶的最新消息 ID（LatestMsgID），如果用戶發(fā)送消息隊(duì)列不為空則退出；
• 3）向 Pi 模塊發(fā)送獲取 N 條消息 ID 的請(qǐng)求 {UIN: uin, StartMsgID: 0, MsgIDNum: N, ExpireFlag: false}，設(shè)置用戶狀態(tài)為 GettingMsgIDList 并等待回應(yīng)；
• 4）根據(jù) Pi 返回的消息 ID 隊(duì)列，向 Xiu 發(fā)起獲取消息請(qǐng)求 {UIN: uin, MsgIDList: msg ID List}，設(shè)置用戶狀態(tài)為 GettingMsgList 并等待回應(yīng)；
• 5）Xiu 返回消息列表后，設(shè)置狀態(tài)為 SendingMsg，并向 Gateway 轉(zhuǎn)發(fā)消息。

所謂 Ack 消息，就是 Broker 經(jīng) Gateway 把消息轉(zhuǎn)發(fā)給 App 后，App 給Broker的消息回復(fù)，告知Broker其最近成功收到消息的 MsgID。?

Ack 消息處理流程如下：
?

• 1）如果用戶狀態(tài)為 OffLine，則退出；
• 2）更新 LatestAckMsgID 的值；
• 3）如果用戶發(fā)送消息隊(duì)列不為空，則發(fā)送下一個(gè)消息后退出；
• 4）如果 LatestAckMsgID >= LatestMsgID，則退出；
• 5）向 Pi 模塊發(fā)送獲取 N 條消息 ID 的請(qǐng)求 {UIN: uin, StartMsgID: 0, MsgIDNum: N, ExpireFlag: false}，設(shè)置用戶狀態(tài)為 GettingMsgIDList 并等待回應(yīng)；
• 6）根據(jù) Pi 返回的消息 ID 隊(duì)列，向 Xiu 發(fā)起獲取消息請(qǐng)求 {UIN: uin, MsgIDList: msg ID List}，設(shè)置用戶狀態(tài)為 GettingMsgList 并等待回應(yīng)；
• 7）Xiu 返回消息列表后，設(shè)置狀態(tài)為 SendingMsg，并向 Gateway 轉(zhuǎn)發(fā)消息。

總體上，PiXiu 轉(zhuǎn)發(fā)消息流程采用拉取（pull）轉(zhuǎn)發(fā)模型，以上面五種消息為驅(qū)動(dòng)進(jìn)行狀態(tài)轉(zhuǎn)換，并作出相應(yīng)的動(dòng)作行為。

九、本文總結(jié)

這套群聊消息系統(tǒng)尚有以下task list需完善：
?

• 1）消息以UDP鏈路傳遞，不可靠【2018/01/29解決之】；
• 2）目前的負(fù)載均衡算法采用了極簡(jiǎn)的RoundRobin算法，可以根據(jù)成功率和延遲添加基于權(quán)重的負(fù)載均衡算法實(shí)現(xiàn)；
• 3）只考慮傳遞，沒(méi)有考慮消息的去重，可以根據(jù)消息ID實(shí)現(xiàn)這個(gè)功能【2018/01/29解決之】；
• 4）各個(gè)模塊之間沒(méi)有考慮心跳方案，整個(gè)系統(tǒng)的穩(wěn)定性依賴于Registry【2018/01/17解決之】；
• 5）離線消息處理【2018/03/03解決之】；
• 6）區(qū)分消息優(yōu)先級(jí)。

網(wǎng)易云信，你身邊的即時(shí)通訊和音視頻技術(shù)專家，了解我們，請(qǐng)戳網(wǎng)易云信官網(wǎng)

想要行業(yè)洞察和技術(shù)干貨，請(qǐng)關(guān)注網(wǎng)易云信博客

本文轉(zhuǎn)載自52im,作者：JackJiang

與50位技術(shù)專家面對(duì)面20年技術(shù)見(jiàn)證，附贈(zèng)技術(shù)全景圖

總結(jié)

以上是生活随笔為你收集整理的一套高可用、易伸缩、高并发的IM群聊架构方案设计实践的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。