消息中間件

應(yīng)用場(chǎng)景

• 異步通信
  ??有些業(yè)務(wù)不想也不需要立即處理消息。消息隊(duì)列提供了異步處理機(jī)制，允許用戶把一個(gè)消息放入隊(duì)列，但并不立即處理它。想向隊(duì)列中放入多少消息就放多少，然后在需要的時(shí)候再去處理它們。
• 解耦
  ??降低工程間的強(qiáng)依賴程度，針對(duì)異構(gòu)系統(tǒng)進(jìn)行適配。在項(xiàng)目啟動(dòng)之初來預(yù)測(cè)將來項(xiàng)目會(huì)碰到什么需求，是極其困難的。通過消息系統(tǒng)在處理過程中間插入了一個(gè)隱含的、基于數(shù)據(jù)的接口層，兩邊的處理過程都要實(shí)現(xiàn)這一接口，當(dāng)應(yīng)用發(fā)生變化時(shí)，可以獨(dú)立的擴(kuò)展或修改兩邊的處理過程，只要確保它們遵守同樣的接口約束。
• 過載保護(hù)
  ??在訪問量劇增的情況下，應(yīng)用仍然需要繼續(xù)發(fā)揮作用，但是這樣的突發(fā)流量無法提取預(yù)知；如果以為了能處理這類瞬間峰值訪問為標(biāo)準(zhǔn)來投入資源隨時(shí)待命無疑是巨大的浪費(fèi)。使用消息隊(duì)列能夠使關(guān)鍵組件頂住突發(fā)的訪問壓力，而不會(huì)因?yàn)橥话l(fā)的超負(fù)荷的請(qǐng)求而完全崩潰。
• 可恢復(fù)性
  ??系統(tǒng)的一部分組件失效時(shí)，不會(huì)影響到整個(gè)系統(tǒng)。消息隊(duì)列降低了進(jìn)程間的耦合度，所以即使一個(gè)處理消息的進(jìn)程掛掉，加入隊(duì)列中的消息仍然可以在系統(tǒng)恢復(fù)后被處理。
• 順序保證
  ??在大多使用場(chǎng)景下，數(shù)據(jù)處理的順序都很重要。大部分消息隊(duì)列本來就是排序的，并且能保證數(shù)據(jù)會(huì)按照特定的順序來處理。
• 緩沖
  ??在任何重要的系統(tǒng)中，都會(huì)有需要不同的處理時(shí)間的元素。消息隊(duì)列通過一個(gè)緩沖層來幫助任務(wù)最高效率的執(zhí)行，該緩沖有助于控制和優(yōu)化數(shù)據(jù)流經(jīng)過系統(tǒng)的速度。以調(diào)節(jié)系統(tǒng)響應(yīng)時(shí)間。
• 數(shù)據(jù)流處理
  ??分布式系統(tǒng)產(chǎn)生的海量數(shù)據(jù)流，如：業(yè)務(wù)日志、監(jiān)控?cái)?shù)據(jù)、用戶行為等，針對(duì)這些數(shù)據(jù)流進(jìn)行實(shí)時(shí)或批量采集匯總，然后進(jìn)行大數(shù)據(jù)分析是當(dāng)前互聯(lián)網(wǎng)的必備技術(shù)，通過消息隊(duì)列完成此類數(shù)據(jù)收集是最好的選擇。

常用協(xié)議

• AMQP協(xié)議
• MQTT協(xié)議
• STOMP協(xié)議
• XMPP協(xié)議
• 其他基于TCP/IP自定義的協(xié)議

常用消息中間件

Kafka
??Apache下的一個(gè)子項(xiàng)目，使用scala實(shí)現(xiàn)的一個(gè)高性能分布式Publish/Subscribe消息隊(duì)列系統(tǒng)，具有以下特性：

• 快速持久化：通過磁盤順序讀寫與零拷貝機(jī)制，可以在O(1)的系統(tǒng)開銷下進(jìn)行消息持久化；
• 高吞吐：在一臺(tái)普通的服務(wù)器上既可以達(dá)到10W/s的吞吐速率；
• 高堆積：支持topic下消費(fèi)者較長(zhǎng)時(shí)間離線，消息堆積量大；
• 完全的分布式系統(tǒng)：Broker、Producer、Consumer都原生自動(dòng)支持分布式，依賴zookeeper自動(dòng)實(shí)現(xiàn)復(fù)雜均衡；
• 支持Hadoop數(shù)據(jù)并行加載：對(duì)于像Hadoop的一樣的日志數(shù)據(jù)和離線分析系統(tǒng)，但又要求實(shí)時(shí)處理的限制，這是一個(gè)可行的解決方案。

RocketMQ
??阿里巴巴的MQ中間件，在其多個(gè)產(chǎn)品下使用，并能夠撐住雙十一的大流量，他并沒有實(shí)現(xiàn)JMS規(guī)范，使用起來很簡(jiǎn)單。部署由一個(gè) 命名服務(wù)（nameserver）和一個(gè)代理（broker）組成，nameserver和broker以及producer都支持集群，隊(duì)列的容量受機(jī)器硬盤的限制，隊(duì)列滿后可以支持持久化到硬盤（也可以自己適配代碼，將其持久化到NOSQL數(shù)據(jù)庫(kù)中），隊(duì)列滿后會(huì)影響吞吐量，可以采用主備來保證穩(wěn)定性，支持回溯消費(fèi)，可以在broker端進(jìn)行消息過濾.

RabbitMQ
??使用Erlang編寫的一個(gè)開源的消息隊(duì)列，本身支持很多的協(xié)議：AMQP，XMPP, SMTP,STOMP，也正是如此，使的它變的非常重量級(jí)，更適合于企業(yè)級(jí)的開發(fā)。同時(shí)實(shí)現(xiàn)了Broker架構(gòu)，核心思想是生產(chǎn)者不會(huì)將消息直接發(fā)送給隊(duì)列，消息在發(fā)送給客戶端時(shí)先在中心隊(duì)列排隊(duì)。對(duì)路由(Routing)，負(fù)載均衡(Load balance)、數(shù)據(jù)持久化都有很好的支持。多用于進(jìn)行企業(yè)級(jí)的ESB整合。結(jié)合erlang語言本身的并發(fā)優(yōu)勢(shì)，性能較好，但是不利于做二次開發(fā)和維護(hù)。

ActiveMQ
??Apache下的一個(gè)子項(xiàng)目。使用Java完全支持JMS1.1和J2EE 1.4規(guī)范的 JMS Provider實(shí)現(xiàn)，少量代碼就可以高效地實(shí)現(xiàn)高級(jí)應(yīng)用場(chǎng)景?？刹灏蔚膫鬏攨f(xié)議支持，比如：in-VM, TCP, SSL, NIO, UDP, multicast, JGroups and JXTA transports。RabbitMQ、ZeroMQ、ActiveMQ均支持常用的多種語言客戶端 C++、Java、.Net,、Python、 Php、 Ruby等。

Redis
??使用C語言開發(fā)的一個(gè)Key-Value的NoSQL數(shù)據(jù)庫(kù)，開發(fā)維護(hù)很活躍，雖然它是一個(gè)Key-Value數(shù)據(jù)庫(kù)存儲(chǔ)系統(tǒng)，但它本身支持MQ功能，所以完全可以當(dāng)做一個(gè)輕量級(jí)的隊(duì)列服務(wù)來使用。對(duì)于RabbitMQ和Redis的入隊(duì)和出隊(duì)操作，各執(zhí)行100萬次，每10萬次記錄一次執(zhí)行時(shí)間。測(cè)試數(shù)據(jù)分為128Bytes、512Bytes、1K和10K四個(gè)不同大小的數(shù)據(jù)。實(shí)驗(yàn)表明：入隊(duì)時(shí)，當(dāng)數(shù)據(jù)比較小時(shí)Redis的性能要高于RabbitMQ，而如果數(shù)據(jù)大小超過了10K，Redis則慢的無法忍受；出隊(duì)時(shí)，無論數(shù)據(jù)大小，Redis都表現(xiàn)出非常好的性能，而RabbitMQ的出隊(duì)性能則遠(yuǎn)低于Redis。

ZeroMQ
??擴(kuò)展性好，開發(fā)比較靈活，采用C語言實(shí)現(xiàn)，實(shí)際上他只是一個(gè)socket庫(kù)的重新封裝，如果我們做為消息隊(duì)列使用，需要開發(fā)大量的代碼，非持久隊(duì)列。

消息中間件對(duì)比

RocketMQ

簡(jiǎn)介

• 能夠保證嚴(yán)格的消息順序
• 提供豐富的消息拉取模式
• 高效的訂閱者水平擴(kuò)展能力
• 實(shí)時(shí)的消息訂閱機(jī)制
• 億級(jí)消息堆積能力

網(wǎng)絡(luò)架構(gòu)圖

架構(gòu)圖

組件特性

nameserver

• nameserver互相獨(dú)立，彼此沒有通信關(guān)系，單臺(tái)nameserver掛掉，不影響其他nameserver，即使全部掛掉，也不影響業(yè)務(wù)系統(tǒng)使用。
• nameserver不會(huì)有頻繁的讀寫，所以性能開銷非常小，穩(wěn)定性很高。
• 集群之間互不通信，降低了nameserver的復(fù)雜程度，對(duì)網(wǎng)絡(luò)的要求也降低了不少，topic路由信息無須在集群之間保持強(qiáng)一致，追求最終一致。

broker

與nameserver關(guān)系

簡(jiǎn)單模型

broker內(nèi)部邏輯圖

集群模型

連接

• 單個(gè)broker和所有nameserver保持長(zhǎng)連接

心跳

• 心跳間隔：每隔30秒（此時(shí)間無法更改）向所有nameserver發(fā)送心跳，心跳包含了自身的topic配置信息
• 心跳超時(shí)：nameserver每隔10秒鐘（此時(shí)間無法更改），掃描所有還存活的broker連接，若某個(gè)連接2分鐘內(nèi)（當(dāng)前時(shí)間與最后更新時(shí)間差值超過2分鐘，此時(shí)間無法更改）沒有發(fā)送心跳數(shù)據(jù)，則斷開連接

斷開

• 時(shí)機(jī)：broker掛掉；心跳超時(shí)導(dǎo)致nameserver主動(dòng)關(guān)閉連接
• 動(dòng)作：一旦連接斷開，nameserver會(huì)立即感知，更新topc與隊(duì)列的對(duì)應(yīng)關(guān)系，但不會(huì)通知生產(chǎn)者和消費(fèi)者

負(fù)載均衡

• 一個(gè)topic分布在多個(gè)broker上，一個(gè)broker可以配置多個(gè)topic，它們是多對(duì)多的關(guān)系
• 如果某個(gè)topic消息量很大，應(yīng)該給它多配置幾個(gè)隊(duì)列，并且盡量多分布在不同broker上，減輕某個(gè)broker的壓力
• topic消息量都比較均勻的情況下，如果某個(gè)broker上的隊(duì)列越多，則該broker壓力越大

可用性

?? 由于消息分布在各個(gè)broker上，一旦某個(gè)broker宕機(jī)，則該broker上的消息讀寫都會(huì)受到影響。所以rocketmq提供了master/slave的結(jié)構(gòu)，salve定時(shí)從master同步數(shù)據(jù)，如果master宕機(jī)，則slave提供消費(fèi)服務(wù)，但是不能寫入消息，此過程對(duì)應(yīng)用透明，由rocketmq內(nèi)部解決。

• 一旦某個(gè)broker master宕機(jī)，生產(chǎn)者和消費(fèi)者多久才能發(fā)現(xiàn)？受限于rocketmq的網(wǎng)絡(luò)連接機(jī)制，默認(rèn)情況下，最多需要30秒，但這個(gè)時(shí)間可由應(yīng)用設(shè)定參數(shù)來縮短時(shí)間。這個(gè)時(shí)間段內(nèi)，發(fā)往該broker的消息都是失敗的，而且該broker的消息無法消費(fèi)，因?yàn)榇藭r(shí)消費(fèi)者不知道該broker已經(jīng)掛掉。
• 消費(fèi)者得到master宕機(jī)通知后，轉(zhuǎn)向slave消費(fèi)，但是slave不能保證master的消息100%都同步過來了，因此會(huì)有少量的消息丟失。但是消息最終不會(huì)丟的，一旦master恢復(fù)，未同步過去的消息會(huì)被消費(fèi)掉。

可用性

可靠性

消息清理

讀寫性能

系統(tǒng)特性

消費(fèi)者

Consumer Group

??用來表示一個(gè)消費(fèi)消息應(yīng)用，一個(gè)Consumer Group下包含多個(gè)Consumer實(shí)例，可以是多臺(tái)機(jī)器，也可以是多個(gè)進(jìn)程，或者是一個(gè)進(jìn)程的多個(gè)Consumer對(duì)象。一個(gè)Consumer Group下的多個(gè)Consumer以均攤方式消費(fèi)消息，如果設(shè)置為廣播方式，那么這個(gè)Consumer Group下的每個(gè)實(shí)例都消費(fèi)全量數(shù)據(jù)。

與nameserver關(guān)系

連接

??單個(gè)消費(fèi)者和一臺(tái)nameserver保持長(zhǎng)連接，定時(shí)查詢topic配置信息，如果該nameserver掛掉，消費(fèi)者會(huì)自動(dòng)連接下一個(gè)nameserver，直到有可用連接為止，并能自動(dòng)重連。

輪詢時(shí)間

??默認(rèn)情況下，消費(fèi)者每隔30秒從nameserver獲取所有topic的最新隊(duì)列情況，這意味著某個(gè)broker如果宕機(jī)，客戶端最多要30秒才能感知。該時(shí)間由DefaultMQPushConsumer的pollNameServerInteval參數(shù)決定，可手動(dòng)配置。

與broker關(guān)系

連接

心跳

斷開

負(fù)載均衡

消費(fèi)機(jī)制

消費(fèi)進(jìn)度存儲(chǔ)

生產(chǎn)者

Producer Group

??用來表示一個(gè)發(fā)送消息應(yīng)用，一個(gè)Producer Group下包含多個(gè)Producer實(shí)例，可以是多臺(tái)機(jī)器，也可以是一臺(tái)機(jī)器的多個(gè)進(jìn)程，或者一個(gè)進(jìn)程的多個(gè)Producer對(duì)象。一個(gè)Producer Group可以發(fā)送多個(gè)Topic消息，Producer Group作用如下：

• 標(biāo)識(shí)一類Producer
• 可以通過運(yùn)維工具查詢這個(gè)發(fā)送消息應(yīng)用下有多個(gè)Producer實(shí)例
• 發(fā)送分布式事務(wù)消息時(shí)，如果Producer中途意外宕機(jī)，Broker會(huì)主動(dòng)回調(diào)Producer Group內(nèi)的任意一臺(tái)機(jī)器來確認(rèn)事務(wù)狀態(tài)。

與nameserver關(guān)系

連接

心跳

輪詢時(shí)間

與broker關(guān)系

連接

心跳

斷開連接

負(fù)載均衡

Broker集群配置方式及優(yōu)缺點(diǎn)

單個(gè) Master

多個(gè) Master

多 Master 多 Slave 模式，異步復(fù)制

多 Master 多 Slave 模式，同步雙寫

演示

單機(jī)演示

啟動(dòng)nameserver

1. 配置jvm參數(shù)

啟動(dòng)broker

1. 配置jvm參數(shù)
2. 配置自動(dòng)創(chuàng)建主題參數(shù)（僅使用測(cè)試環(huán)境，生產(chǎn)環(huán)境不建議）
3. 手動(dòng)創(chuàng)建主體（命令或者直接修改store的配置文件，需要重啟）

啟動(dòng)生產(chǎn)者

啟動(dòng)消費(fèi)者

管控臺(tái)演示

集群搭建及演示

源碼構(gòu)建

構(gòu)建rocketmq

git clone https://github.com/apache/rocketmq.git
mvn -Prelease-all -DskipTests clean install -U
cd distribution/target/apache-rocketmq
nohup sh bin/mqbroker -n localhost:9876 autoCreateTopicEnable=true &

構(gòu)建管控臺(tái)：

注意：pom文件中，把<rocketmq.version>4.4.0</rocketmq.version>修改，4.4.0在maven中心倉(cāng)庫(kù)中有，自己本地可以安裝rocketmq最新代碼的快照版本，但是仍然會(huì)有問題

git clone https://github.com/apache/rocketmq-externals.git
cd rocketmq-console
mvn clean package -Dmaven.test.skip=true
java -jar target/rocketmq-console-ng-1.0.0.jar