張樂

摘 要 針對現(xiàn)在移動互聯(lián)網(wǎng)復(fù)雜、信號不穩(wěn)定的網(wǎng)絡(luò)特點，設(shè)計了一套以 RabbitMQ消息中間件的即時通訊系統(tǒng)，在確保服務(wù)質(zhì)量的情況下，減少了消息的冗余，減少了應(yīng)用間的耦合關(guān)系。使移動設(shè)備更加省電省流量，適用于當(dāng)今移動互聯(lián)網(wǎng)。

【關(guān)鍵詞】AMQP RabbitMQ 即時通訊

1 引言

近幾年來，移動互聯(lián)網(wǎng)憑借其攜帶方便、接入迅速、業(yè)務(wù)內(nèi)容豐富等特點，取得了前所未有的高速發(fā)展。目前，移動應(yīng)用服務(wù)呈現(xiàn)多樣化，但是移動終端最基本的功能是滿足用戶的溝通需求，因此即時通訊類應(yīng)用以其隨時隨地溝通的特點，滿足了用戶的需求。

目前常用于即時通訊的兩大協(xié)議是XMPP協(xié)議和SIMPLE協(xié)議。XMPP和SIMPLE 都是根據(jù)消息體里的消息頭尋址的，在消息體里面含有消息的發(fā)送者、接收者、消息路由用的會話標(biāo)示信息等眾多頭域，使得消息體的體積變大，帶寬利用率較低。XMPP 和 SIMPLE協(xié)議都是基于字符文本的通信協(xié)議，其優(yōu)點是可讀性強，便于抓包分析，但字符文本協(xié)議通信效率較低，并且，為了保證通信安全，采用TLS 等加密傳輸計算量也較大，耗能較高。

移動互聯(lián)網(wǎng)相對傳統(tǒng)互聯(lián)網(wǎng)最顯著的特點就是其移動性，這種移動性隨之帶來的是諸多不穩(wěn)定、不可靠和隨意性，使得移動互聯(lián)網(wǎng)上的應(yīng)用與應(yīng)用服務(wù)器之間建立的網(wǎng)絡(luò)連接可靠性較差，很難保持長時間的連接狀態(tài)，因此傳統(tǒng)的通信模型，比如 SIMPLE和XMPP 中采用的消息傳遞，不適合應(yīng)用于移動互聯(lián)網(wǎng)，移動互聯(lián)網(wǎng)應(yīng)用與應(yīng)用服務(wù)器之間可以通過松散耦合的關(guān)系來改善移動性帶來的問題。

2 AMQP及RabbitMQ

2.1 AMQP

AMQP（AdvancedMessage Queuing Protocol），即高級消息隊列是一個基于消息異步處理的應(yīng)用層高級消息隊列協(xié)議，是消息中間件的開發(fā)標(biāo)準(zhǔn)。它的主要特征是面向消息、隊列、路由，且安全、可靠。AMQP是基于客戶端/代理模式，為客戶端應(yīng)用與消息中間件之間提供異步、安全、高效的交互?；诖藚f(xié)議的客戶端與消息中間件可傳遞消息，并不受客戶端和中間件不同產(chǎn)品、不同開發(fā)語言等條件的限制。

2.2 RabbitMQ

RabbitMQ是流行的開源消息隊列系統(tǒng)，由以高性能、健壯以及可伸縮性出名的erlang 語言開發(fā)，是 AMQP 的標(biāo)準(zhǔn)實現(xiàn)。它可以支持各種消息交換的體系結(jié)構(gòu)：

（1）存儲轉(zhuǎn)發(fā)（多個消息發(fā)送者，單個消息接收者）；

（2）分布式事務(wù)（多個消息發(fā)送者，多個消息接收者）；

（3）發(fā)布訂閱（多個消息發(fā)送者，多個消息接收者）；

（4）基于內(nèi)容的路由（多個消息發(fā)送者，多個消息接收者）；

（5）文件傳輸隊列（多個消息發(fā)送者，多個消息接收者）；

（6）點對點連接（單個消息發(fā)送者，單個消息接收者）。

RabbitMQ的主要特性包括如下方面：

（1）定義、隊列、通道等概念，使對消息投遞的控制更加精準(zhǔn)靈活；

（2）支持高可用，以及消息和隊列的持久化；

（3）支持集群，允許節(jié)點動態(tài)變化和節(jié)點故障；

（4）支持眾多語言的客戶端，包括Java、C/C++、php、.Netpython等；

（5）高性能、消息吞吐量大。

生產(chǎn)者發(fā)送消息到RabbitMQ 服務(wù)器，在服務(wù)器內(nèi)部，用戶創(chuàng)建交換機和隊列，通過綁定規(guī)則將兩者聯(lián)系在一起。交換機負責(zé)分發(fā)消息，根據(jù)類型綁定的不同分發(fā)策略有區(qū)別。消息最后來到隊列中，等待消費者取走。

3 即時通訊系統(tǒng)的設(shè)計

本文提出一種以RabbitMQ為消息中間件，采用發(fā)布/訂閱模型作為消息傳輸機制，以protobuf作為數(shù)據(jù)傳輸格式。采用C/S 體系結(jié)構(gòu)，將消息存儲與轉(zhuǎn)發(fā)，確保消息不重不漏，充分的適應(yīng)移動互聯(lián)網(wǎng)的一套即時通訊的方案。

本文的消息系統(tǒng)可以看做是一個發(fā)布/訂閱系統(tǒng)，有消息生產(chǎn)者publisher，消息中轉(zhuǎn)者RabbitMQ Server，消息處理者Message Server，以及消息消費者Receiver.這個系統(tǒng)中的publisher和receiver共同構(gòu)成了一個channel。

4 消息流程

4.1 發(fā)送流程

如圖1所示 。

（1）客戶端A發(fā)送一條消息內(nèi)容為send_client_uId【發(fā)送者id】 ， channel name ，msg time【消息發(fā)送時間，精確到秒】，msg content【消息內(nèi)容】，保存在本地的SqliteSQL數(shù)據(jù)庫，然后用protobuf序列化后發(fā)發(fā)個RabbitMQ。

（2）Message服務(wù)器的消息隊列通過RabbitMQ收到來自客戶端A的消息，反序列化。

（3）Message服務(wù)器收到消息后以channel name+ msg time為key到本地消息緩存中查詢消息是否已經(jīng)存在，如果存在則終止消息流程，通過RabbitMQ服務(wù)器發(fā)送"duplicate msg"這個msg ack 給客戶端A，否則繼續(xù)。

（4）Message服務(wù)器到Counter服務(wù)器（消息計數(shù)器，為每個text等類型的消息分配msg id）以channel name為key查詢其最新的msg id，把msg id自增一后作為這條消息的id。

（5）Message服務(wù)器把分配好id的消息插入本地msg cache和msg DB中。

（6）Message服務(wù)器給客戶端A返回ack， 內(nèi)容為msg id ， msg time ， channel name。

（7）客戶端A收到ack包后終止消息流程，并刪除本地數(shù)據(jù)庫中的數(shù)據(jù)。如果在發(fā)送流程超時后仍未收到消息則轉(zhuǎn)到步驟1進行重試，并計算重試次。

（8）如果重試次數(shù)超過兩次依然失敗則提示“系統(tǒng)繁忙” or “網(wǎng)絡(luò)環(huán)境不佳，請稍后再嘗試發(fā)送”等，終止消息發(fā)送流程。

4.2 接收流程

Message服務(wù)器到RabbitMQ中檢測Channel中的receive_client是否在線，如果在線則將消息發(fā)送出去。

4.3 心跳發(fā)送流程

（1）客戶端發(fā)送心跳包，內(nèi)容為{client_uId， network type， list{channel name：newest channel msg id} }，即心跳包要上報客戶端所在的所有channel，以及本地歷史消息記錄中每個channel最新的消息的id；心跳包轉(zhuǎn)給專門處理心跳邏輯的心跳服務(wù)器。

（2）心跳服務(wù)器收到心跳包后到Counter服務(wù)器循環(huán)查詢每個channel的最新消息id，如果客戶端上報的id與這個id不等，就發(fā)送一條消息通知Message服務(wù)器，消息內(nèi)容為{publish_uId， channel name， client newest msg id of channel【channel內(nèi)的最新消息Id】}。

（3）Message服務(wù)器收到這條消息后，重新啟動消息下發(fā)邏輯，到緩存中取出所有的大于{client newest msg id of channel}的id列表。

（4）Message服務(wù)器依據(jù)list中的id到消息存儲服務(wù)器中依次取出每條消息。

（5）Message服務(wù)器把這些消息作為"未讀消息"下發(fā)給客戶端。

（6）心跳服務(wù)器給客戶端下發(fā)heartbeat ack包，數(shù)據(jù)包括其所在的每個channel的最新消息的msg id。

（7）客戶端收到heartbeat ack包后，依據(jù)每個channel的最新的msg id與本地消息緩存中對應(yīng)的channel的最新消息id做對比，如果id不等，客戶端可以啟動拉取消息流程。

5 結(jié)束語

本文簡單介紹了基于AMQP協(xié)議的標(biāo)準(zhǔn)實現(xiàn)RabbitMQ，并提出了以開源跨平臺的RabbitMQ為消息中間件，松耦合的發(fā)布/訂閱模型為通訊方式，protocolBuffer為數(shù)據(jù)傳輸格式的一套完備即時通訊系統(tǒng)設(shè)計，詳細敘述了消息下發(fā)的技術(shù)流程，同時保證了消息不重不漏。從根本上解決了由于消息冗余而占用大量的帶寬利用率，使得該方案更加適合移動互聯(lián)網(wǎng)復(fù)雜的網(wǎng)絡(luò)環(huán)境。下一步的工作將著重研究在復(fù)雜的移動網(wǎng)絡(luò)下，用于保持連接的心跳包的發(fā)送時間間隔的方案和消息傳遞時的安全問題。

參考文獻

[1]王默涵.面向移動互聯(lián)網(wǎng)的Presence/IM機制的設(shè)計與實現(xiàn)[D].小型微型計算機系統(tǒng)，2015（04）.

[2]高曉婷.基于AMQP的信息發(fā)布與訂閱[D].2013（10）.

[3]RabbitMQ與AMQP協(xié)議詳解.http：//www.cnblogs.com/frankyou/p/5283539.html

[4]Rabbit MQ[EB/OL].http：//www.rabbitmq.com/.z