《RocketMQ實(shí)戰(zhàn)與原理解析》

消息重復(fù)問(wèn)題

? 對(duì)分布式消息隊(duì)列來(lái)說(shuō)椒舵，同時(shí)做到確保一定投遞和不重復(fù)投遞是很難的蚂踊， 也就是所謂的“有且僅有 1 次” 在魚(yú)和熊掌不可兼得的情況下， RocketMQ 選擇了確保一定投遞成功笔宿，保證消息不丟失犁钟，但有可能造成消息重復(fù)。

? 消息重復(fù)一般情況下不會(huì)發(fā)生泼橘，但是如果消息量大涝动，網(wǎng)絡(luò)有波動(dòng)，消息重復(fù)就是個(gè)大概率事件炬灭。比如 Producer 有個(gè)函數(shù) setRetryTimesWhenSendFailed, 設(shè)置在同步方式下自動(dòng)重試的次數(shù)醋粟，默認(rèn)值是 2 ，這樣當(dāng)?shù)谝淮伟l(fā)送消息時(shí)担败， Broker 端接收到了消息但是沒(méi)有正確返回發(fā)送成功的狀態(tài)昔穴，就造成了消息重復(fù)。

? 解決消息重復(fù)有兩種方法：第一種方法是保證消費(fèi)邏輯的幕等性（多次調(diào)用和一次調(diào)用效果相同）提前；另一種方法是維護(hù)一個(gè)巳消費(fèi)消息的記錄吗货，消費(fèi)前查詢(xún)這個(gè)消息是否被消費(fèi)過(guò)這兩種方法都需要使用者自己實(shí)現(xiàn)。

動(dòng)態(tài)增減機(jī)器

? 一個(gè)消息隊(duì)列集群由 多臺(tái)機(jī)器組成狈网，持續(xù)穩(wěn)定地提供服務(wù)宙搬，因?yàn)闃I(yè)務(wù)需求或硬件故障，經(jīng)常需要增加或減少各個(gè)角色的機(jī)器拓哺，本節(jié)介紹如何在不影響服務(wù)穩(wěn)定性的情況下動(dòng)態(tài)地增減機(jī)器勇垛。

動(dòng)態(tài)增減 NameServer

? NameServer 是 RocketMQ 集群的協(xié)調(diào)者，集群的各個(gè)組件是通過(guò) NameServer 獲取各種屬性和地址信息的士鸥。主要功能包括兩部分：一個(gè)是各個(gè) Broker 定期上報(bào)自己的狀態(tài)信息到 NameServer 闲孤；另一個(gè)是各個(gè)客戶(hù)端 ，包括 Producer 和 Consumer 烤礁，以及命令行工具讼积，通過(guò) NameServer 獲取最新的狀態(tài)信。所以脚仔，在啟動(dòng) Broker 勤众、生產(chǎn)者和消費(fèi)者之前，必須告訴它們 NameServer 的地址鲤脏，為了提高可靠性们颜，建議啟動(dòng)多個(gè) NameServer吕朵。NameServer 占用資源不多，可以和 Broker 部署在同一臺(tái)機(jī)器窥突。有多個(gè) NameServer 后努溃，減少某個(gè) NameServer 不會(huì)對(duì)其他組件產(chǎn)生影響。

? 有四種種方式可設(shè)置 NameServer 的地址波岛， 下面按優(yōu)先級(jí)由高到低依次介紹：

? 1 ）通過(guò)代碼設(shè)置茅坛，比如在 Producer 中，通過(guò) Producer.setNamesrvAddr (”name-server1-ip:port;name-server2-ip:port”) 來(lái)設(shè) mqadmin 命令行工具中则拷，是通過(guò)-n name-server-ip1:port;name-server-ip2:port 參數(shù)來(lái)設(shè)置的 如果自 定義了命令行工具，也可以通過(guò) defaultMQAdminExt.setNamesrvAddr(” nameserver1-ip：port;name-server2-ipport”) 來(lái)設(shè)置的曹鸠。

? 2 ）使用 Java 啟動(dòng)參數(shù)設(shè)置煌茬，對(duì)應(yīng) option 是 rocketmq.namesrv.addr。

? 3 ）通過(guò) Linux 環(huán)境變量設(shè)置彻桃，在啟動(dòng)前設(shè)置變量： NAMESRV_ADDR坛善。

? 4 ）通過(guò) HTTP 服務(wù)來(lái)設(shè)置，當(dāng)上述方法都沒(méi)有使用邻眷，程序會(huì)向一個(gè) HTTP 地址發(fā)送請(qǐng)求來(lái)獲取 NameServer 地址眠屎，默認(rèn)的 URL http://jmenv.tbsite.net:8080/rocketmq/nsaddr （淘寶的測(cè)試地址），通過(guò) rocketmq.namesrv.domain 參數(shù)來(lái)覆蓋 jmenv.tbsite.net 肆饶；通過(guò) rocketmq.namesrv.domain.subgroup 參數(shù)來(lái)覆蓋 nsaddr 種方式看似繁瑣改衩，但它是唯一支持動(dòng)態(tài)增加 NameServer ，無(wú)須重啟其他組件的方式驯镊。使用這種方式后其他組件會(huì)每隔分鐘請(qǐng)求一次該 URL 葫督，獲取最新的 NameServer 地址。

動(dòng)態(tài)增減 Broker

? 由于業(yè)務(wù)增長(zhǎng),需要對(duì)集群進(jìn)行擴(kuò)容的時(shí)候,可以動(dòng)態(tài)增加 Broker 角色的機(jī)器板惑。只增加 Broker 不會(huì)對(duì)原有的 Topic 產(chǎn)生影響,原來(lái)創(chuàng)建好的 Topic 中數(shù)據(jù)的讀寫(xiě)依然在原來(lái)的那些 Broker上進(jìn)行橄镜。

? 集群擴(kuò)容后,一是可以把新建的 Topic 指定到新的 Broker 機(jī)器上,均衡利用資源;另一種方式是通過(guò) updateTopic 命令更改現(xiàn)有的 Topic 配置,在新加的 Broker 上創(chuàng)建新的隊(duì)列。比如 TestTopic 是現(xiàn)有的一個(gè) Topic ,因?yàn)閿?shù)據(jù)量增大需要擴(kuò)容,新增的一個(gè) Broker機(jī)器地址是 192.168.0.1:10911 ,這個(gè)時(shí)候執(zhí)行下面的命令: sh ./bin/mqadmin updateTopic -b 192.168.0.1:1091 -t TestTopic -n 192.168.0.100:9876 ,結(jié)果是在新增的 Broker機(jī)器上,為 TestTopic 新創(chuàng)建了 8 個(gè)讀寫(xiě)隊(duì)列冯乘。

? 如果因?yàn)闃I(yè)務(wù)變動(dòng)或者置換機(jī)器需要減少 Broker ,此時(shí)該如何操作呢?減少 Broker 要看是否有持續(xù)運(yùn)行的 Producer ,當(dāng)一個(gè) Topic 只有一個(gè) Master Broker,停掉這個(gè) Broker后,消息的發(fā)送肯定會(huì)受到影響,需要在停止這個(gè) Broker 前,停止發(fā)送消息洽胶。

? 當(dāng)某個(gè) Topic 有多個(gè) Master Broker,停了其中一個(gè),這時(shí)候是否會(huì)丟失消息呢? 答案和 Producer 使用的發(fā)送消息的方式有關(guān),如果使用同步方式 send(msg) 發(fā)送,在 DefaultMQProducer內(nèi)部有個(gè)自動(dòng)重試邏輯,其中個(gè) Broker停了,會(huì)自動(dòng)向另一個(gè) Broker發(fā)消息,不會(huì)發(fā)生丟消息現(xiàn)象。如果使用異步方式發(fā)送send(msg, callback),或者用 sindOne Way方式,會(huì)丟失切換過(guò)程中的消息裆馒。因?yàn)樵诋惒胶?sindOneWay 這兩種發(fā)送方式下 Producer.setRetryTimesWhenSendFailed 設(shè)置不起作用,發(fā)送失敗不會(huì)重試 DefaultMQProducer 默認(rèn)每30秒到 Name Server 請(qǐng)求最新的路由消息, Producer 如果獲取不到已停止的 Broker下的隊(duì)列信息,后續(xù)就自動(dòng)不再向這些隊(duì)列發(fā)送消息姊氓。

? 如果 Producer 程序能夠暫停,在有一個(gè) Master 和一個(gè) Slave 的情況下也可以順利切換×熳罚可以關(guān)閉 Producer后關(guān)閉 Master Broker,這個(gè)時(shí)候所有的讀取都會(huì)被定向到 Slave機(jī)器,消費(fèi)消息不受影響他膳。把 Master Broker機(jī)器置換完后基于原來(lái)的數(shù)據(jù)啟動(dòng)這個(gè) Master Broker,然后再啟動(dòng) Producer 程序正常發(fā)送消息。

? 用 Linux的 kill pid 命令就可以正確地關(guān)閉 Broker, BrokerController下有個(gè) shutdown 函數(shù),這個(gè)函數(shù)被加到了 Shutdown Hook里,當(dāng)用 Linux 的 klt 命令時(shí)(不能用 kll -9 ), shutdown函數(shù)會(huì)先被執(zhí)行绒窑。也可以通過(guò) RocketMQ 提供的工具 ( mqshutdown broker ) 來(lái)關(guān)閉 Broker,它們的原理是一樣的棕孙。

各種故障對(duì)消息的影響

? 我們期望消息隊(duì)列集群一直可靠穩(wěn)定地運(yùn)行,但有時(shí)候故障是難免的,本節(jié)我們列出可能的故障情況,看看如何處理 :

? 1) Broker正常關(guān)閉,啟動(dòng);

? 2) Broker異常 Crash,然后啟動(dòng);

? 3) OS Crash,重啟;

? 4)機(jī)器斷電,但能馬上恢復(fù)供電

? 5)磁盤(pán)損壞;

? 6)CPU、主板、內(nèi)存等關(guān)鍵設(shè)備損壞蟀俊。

? 假設(shè)現(xiàn)有的 RocketMQ 集群,每個(gè) Topic 都配有多 Master 角色的 Broker 供入,并且每個(gè) Master都至少有一個(gè) Slave 機(jī)器(用兩臺(tái)物理機(jī)就可以實(shí)現(xiàn)上述配置),我們來(lái)看看在上述情況下消息的可靠性情況钦铺。

? 第1種情況屬于可控的軟件問(wèn)題,內(nèi)存中的數(shù)據(jù)不會(huì)丟失。如果重啟過(guò)程中有持續(xù)運(yùn)行的 Consumer , Master 機(jī)器出故障后, Consumer 會(huì)自動(dòng)重連到對(duì)應(yīng)的 Slave 機(jī)器,不會(huì)有消息丟失和偏差肢预。當(dāng) Master角色的機(jī)器重啟以后, Consumer 又會(huì)重新連接到 Master 機(jī)器 (注意在啟動(dòng) Master機(jī)器的時(shí)候, 如果 Consumer 正在從 Slave 消費(fèi)消息,不要停止 Consumer矛洞。假如此時(shí)先停止 Consumer 后再啟動(dòng) Master 機(jī)器,然后再啟動(dòng) Consumer ,這個(gè)時(shí)候 Consumer 就會(huì)去讀 Master 機(jī)器上已經(jīng)滯后的 offset值,造成消息大量重復(fù))。

? 如果第1種情況出現(xiàn)時(shí)有持續(xù)運(yùn)行的 Producer,一臺(tái) Master出故障后 Producer 只能向 Topic 下其他的 Master 機(jī)器發(fā)送消息,如果 Producer 采用同步發(fā)送方式,不會(huì)有消息丟失烫映。

? 第2沼本、3、4種情況屬于軟件故障,內(nèi)存的數(shù)據(jù)可能丟失,所以刷盤(pán)策略不同,造成的影響也不同,如果 Master锭沟、 Slave 都配置成 SYNC_FLUSH,可以達(dá)到和第1種情況相同的效果抽兆。

? 第5、6種情況屬于硬件故障,發(fā)生第5族淮、6種情況的故障,原有機(jī)器的磁盤(pán)數(shù)據(jù)可能會(huì)丟失辫红。如果 Master和 Slave機(jī)器間配置成同步復(fù)制方式,某一臺(tái)機(jī)器發(fā)生 5 或 6 的故障,也可以達(dá)到消息不丟失的效果。如果 Master 和 Slave 機(jī)器間是異步復(fù)制,兩次Sync間的消息會(huì)丟失祝辣。

? 總的來(lái)說(shuō),當(dāng)設(shè)置成 :

? 1) 多 Master,每個(gè) Master帶有 Slave;

? 2) 主從之間設(shè)置成 SYNC_MASTER;

? 3) Producer 用同步方式寫(xiě);

? 4) 刷盤(pán)策略設(shè)置成 SYNC_FLUSH;

就可以消除單點(diǎn)依賴(lài)贴妻，即使某臺(tái)機(jī)器出現(xiàn)極端故障也不會(huì)丟消息。

消息的優(yōu)先級(jí)

? 有些場(chǎng)景蝙斜，需要應(yīng)用程序處理幾種類(lèi)型的消息名惩，不同消息的優(yōu)先級(jí)不同。RocketMQ 是個(gè)先進(jìn)先出的隊(duì)列乍炉，不支持消息級(jí)別或者 Topic 級(jí)別的優(yōu)先級(jí)绢片。業(yè)務(wù)中簡(jiǎn)單的優(yōu)先級(jí)需求，可以通過(guò)間接的方式解決岛琼，下面列舉三種優(yōu)先級(jí)相關(guān)需求的具體處理方法底循。

? 第一種是比較簡(jiǎn)單的情況，如果當(dāng)前 Topic 里有多種相似類(lèi)型的消息槐瑞，比如類(lèi)型 AA熙涤、AB、AC 困檩，當(dāng) AB祠挫、AC 的消息量很大，但是處理速度比較慢的時(shí)候悼沿， 隊(duì)列里會(huì)有很多 AB等舔、AC 類(lèi)型的消息在等候處理，這個(gè)時(shí)候如果有少量 AA 型的消息加人糟趾，就會(huì)排在 AB慌植、AC 類(lèi)型消息后面甚牲，需要等候很長(zhǎng)時(shí)間才能被處理。如果業(yè)務(wù)需要 AA 類(lèi)型的消息被及時(shí)處理蝶柿，可以把這三種相似類(lèi)型的消息 分拆到兩個(gè) Topic 里丈钙，比如 AA 類(lèi)型的消息在一個(gè)單獨(dú)的 Topic, AB、AC 類(lèi)型 的消息在另外一個(gè) Topic 把消息分到兩個(gè) Topic 中以后交汤，應(yīng)用程序創(chuàng)建兩個(gè) Consumer 雏赦，分別訂閱不同的 Topic ，這樣消息 AA 在單獨(dú)的 Topic 里芙扎，不會(huì)因 AB星岗、AC 類(lèi)型的消息太多而被長(zhǎng)時(shí)間延時(shí)處理。

? 第二種情況和第一種情況類(lèi)似戒洼，但是不用創(chuàng)建大量的 Topic 伍茄。舉個(gè)實(shí)際應(yīng)用場(chǎng)景：一個(gè)訂單處理系統(tǒng)，接收從 100 家快遞門(mén)店過(guò)來(lái)的請(qǐng)求施逾，把這些請(qǐng)求 通過(guò) Producer 寫(xiě)人 RocketMQ ；訂單處理程序通過(guò) Consumer 從隊(duì)列里讀取消息并處理例获，每天最多處理 1 萬(wàn)單汉额。如果這 100 個(gè)快遞門(mén)店中某幾個(gè)門(mén)店訂單量大增，比如門(mén)店一接了個(gè)大客戶(hù)榨汤，一個(gè)上午就發(fā)出 1 萬(wàn)單消息請(qǐng)求蠕搜，這樣其他 99 家門(mén)店可能被迫等待門(mén)店一的 1 萬(wàn)單處理完，也就是兩天后訂單才能被處理收壕，顯然很不公平妓灌。這時(shí)可以創(chuàng)建一個(gè) Topic ,設(shè)置 Topic 的 MessageQueue 數(shù)量超過(guò)100個(gè), Producer根據(jù)訂單的門(mén)店號(hào),把每個(gè)門(mén)店的訂單寫(xiě)人一個(gè) MessageQueue。DefaultMQPushConsumer 默認(rèn)是采用循環(huán)的方式逐個(gè)讀取一個(gè) Topic 的所有 MessageQueue ,這樣如果某家門(mén)店訂單量大增,這家門(mén)店對(duì)應(yīng)的 MessageQueue 消息數(shù)增多,等待時(shí)間增長(zhǎng),但不會(huì)造成其他家門(mén)店等待時(shí)間增長(zhǎng)蜜宪。DefaultMQPushConsumer 默認(rèn)的 pullBatchSize是32,也就是每次從某個(gè) MessageQueue 讀取消息的時(shí)候,最多可以讀32個(gè)虫埂。在上面的場(chǎng)景中,為了更加公平,可以把 pullBatchSize設(shè)置成 1。

? 第三種情況是強(qiáng)優(yōu)先級(jí)需求,上兩種情況對(duì)消息的“優(yōu)先級(jí)”要求不高,更像一個(gè)保證公平處理的機(jī)制,避免某類(lèi)消息的增多阻塞其他類(lèi)型的消息∑匝椋現(xiàn)在有一個(gè)應(yīng)用程序同時(shí)處理 TypeA掉伏、TypeB、TypeC 三類(lèi)消息澳窑。 TypeA 處于第一優(yōu)先級(jí),要確保只要有 TypeA 消息,必須優(yōu)先處理; TypeB 處于第二優(yōu)先級(jí); TypeC 處于第三優(yōu)先級(jí)斧散。對(duì)這種要求,或者邏輯更復(fù)雜的要求,就要用戶(hù)自己編碼實(shí)現(xiàn)優(yōu)先級(jí)控制,如果上述的三類(lèi)消息在一個(gè) Topic里,可以使用 PullConsumer ,自主控制 MessageQueue 的遍歷,以及消息的讀取;如果上述三類(lèi)消息在三個(gè) Topic 下,需要啟動(dòng)三個(gè) Consumer ,實(shí)現(xiàn)邏輯控制三個(gè) Consumer 的消費(fèi)。