消息隊列之如何保證消息的可靠傳輸
關(guān)注我們,設(shè)為星標(biāo),每天7:30不見不散,架構(gòu)路上與您共享 回復(fù)"架構(gòu)師"獲取資源
1.1 RQ
對于RQ來說在消息傳輸?shù)倪^程中主要有三塊地方可能發(fā)生消息丟失：
生產(chǎn)者
RQ
消費(fèi)者
1.1.1 RQ消費(fèi)者發(fā)生消息丟失
此時可以選擇用 RabbitMQ 提供的事務(wù)功能，就是生產(chǎn)者發(fā)送數(shù)據(jù)之前開啟 RabbitMQ 事務(wù) channel.txSelect ，然后發(fā)送消息，如果消息沒有成功被 RabbitMQ 接收到，那么生產(chǎn)者會收到異常報錯，此時就可以回滾事務(wù) channel.txRollback ，然后重試發(fā)送消息；如果收到了消息，那么可以提交事務(wù) channel.txCommit 。
// 開啟事務(wù)
channel.txSelect
try {
// 這里發(fā)送消息
} catch (Exception e) {
channel.txRollback
// 這里再次重發(fā)這條消息
}
// 提交事務(wù)
channel.txCommit
但是問題是，RabbitMQ 事務(wù)機(jī)制（同步）一搞，基本上吞吐量會下來，因為太耗性能。
所以一般來說，如果你要確保說寫 RabbitMQ 的消息別丟，可以開啟 confirm 模式，在生產(chǎn)者那里設(shè)置開啟 confirm 模式之后，你每次寫的消息都會分配一個唯一的 id，然后如果寫入了 RabbitMQ 中，RabbitMQ 會給你回傳一個 ack 消息，告訴你說這個消息 ok 了。如果 RabbitMQ 沒能處理這個消息，會回調(diào)你的一個 nack 接口，告訴你這個消息接收失敗，你可以重試。而且你可以結(jié)合這個機(jī)制自己在內(nèi)存里維護(hù)每個消息 id 的狀態(tài)，如果超過一定時間還沒接收到這個消息的回調(diào)，那么你可以重發(fā)。
事務(wù)機(jī)制和 confirm 機(jī)制最大的不同在于，事務(wù)機(jī)制是同步的，你提交一個事務(wù)之后會阻塞在那兒，但是 confirm 機(jī)制是異步的，你發(fā)送個消息之后就可以發(fā)送下一個消息，然后那個消息 RabbitMQ 接收了之后會異步回調(diào)你的一個接口通知你這個消息接收到了。
所以一般在生產(chǎn)者這塊避免數(shù)據(jù)丟失，都是用 confirm 機(jī)制的。
1.1.2 RQ本身出問題宕機(jī)
就是 RabbitMQ 自己弄丟了數(shù)據(jù)，這個你必須開啟 RabbitMQ 的持久化，就是消息寫入之后會持久化到磁盤，哪怕是 RabbitMQ 自己掛了，恢復(fù)之后會自動讀取之前存儲的數(shù)據(jù)，一般數(shù)據(jù)不會丟。除非極其罕見的是，RabbitMQ 還沒持久化，自己就掛了，可能導(dǎo)致少量數(shù)據(jù)丟失，但是這個概率較小。
設(shè)置持久化有兩個步驟：
創(chuàng)建 queue 的時候?qū)⑵湓O(shè)置為持久化
這樣就可以保證 RabbitMQ 持久化 queue 的元數(shù)據(jù)，但是它是不會持久化 queue 里的數(shù)據(jù)的。
第二個是發(fā)送消息的時候?qū)⑾⒌?deliveryMode 設(shè)置為 2
就是將消息設(shè)置為持久化的，此時 RabbitMQ 就會將消息持久化到磁盤上去。
必須要同時設(shè)置這兩個持久化才行，RabbitMQ 哪怕是掛了，再次重啟，也會從磁盤上重啟恢復(fù) queue，恢復(fù)這個 queue 里的數(shù)據(jù)。
注意，哪怕是你給 RabbitMQ 開啟了持久化機(jī)制，也有一種可能，就是這個消息寫到了 RabbitMQ 中，但是還沒來得及持久化到磁盤上，結(jié)果不巧，此時 RabbitMQ 掛了，就會導(dǎo)致內(nèi)存里的一點點數(shù)據(jù)丟失。
所以，持久化可以跟生產(chǎn)者那邊的 confirm 機(jī)制配合起來，只有消息被持久化到磁盤之后，才會通知生產(chǎn)者 ack 了，所以哪怕是在持久化到磁盤之前，RabbitMQ 掛了，數(shù)據(jù)丟了，生產(chǎn)者收不到 ack ，你也是可以自己重發(fā)的。
1.1.3 消費(fèi)者丟失消息
RabbitMQ 如果丟失了數(shù)據(jù)，主要是因為你消費(fèi)的時候，剛消費(fèi)到，還沒處理，結(jié)果進(jìn)程掛了，比如重啟了，那么就尷尬了，RabbitMQ 認(rèn)為你都消費(fèi)了，這數(shù)據(jù)就丟了。
這個時候得用 RabbitMQ 提供的 ack 機(jī)制，簡單來說，就是你必須關(guān)閉 RabbitMQ 的自動 ack ，可以通過一個 api 來調(diào)用就行，然后每次你自己代碼里確保處理完的時候，再在程序里 ack 一把。這樣的話，如果你還沒處理完，不就沒有 ack 了？那 RabbitMQ 就認(rèn)為你還沒處理完，這個時候 RabbitMQ 會把這個消費(fèi)分配給別的 consumer 去處理，消息是不會丟的。
1.1.4 RQ總結(jié)
2.1 Kafka
跟RQ一樣，Kafka發(fā)生消息丟失的地方也大致是這三個地方：生產(chǎn)者，Kafka,消費(fèi)者。
2.1.1 Kafka生產(chǎn)者
一般我們?yōu)榱朔胖a(chǎn)者丟失消息一般在生產(chǎn)環(huán)境中這樣設(shè)置：
在 producer 端設(shè)置 acks=all ：這個是要求每條數(shù)據(jù)，必須是寫入所有 replica 之后，才能認(rèn)為是寫成功了。
在 producer 端設(shè)置 retries=MAX （很大很大很大的一個值，無限次重試的意思）：這個是要求一旦寫入失敗，就無限重試，卡在這里了。
這樣就可以保證一定不會丟，因為要求的是：你的 leader 接收到消息，所有的 follower 都同步到了消息之后，才認(rèn)為本次寫成功了。如果沒滿足這個條件，生產(chǎn)者會自動不斷的重試，重試無限次。
2.1.2 Kafka弄丟了消息
這塊比較常見的一個場景，就是 Kafka 某個 broker 宕機(jī)，然后重新選舉 partition 的 leader。大家想想，要是此時其他的 follower 剛好還有些數(shù)據(jù)沒有同步，結(jié)果此時 leader 掛了，然后選舉某個 follower 成 leader 之后，不就少了一些數(shù)據(jù)？這就丟了一些數(shù)據(jù)啊。
生產(chǎn)環(huán)境也遇到過，我們也是，之前 Kafka 的 leader 機(jī)器宕機(jī)了，將 follower 切換為 leader 之后，就會發(fā)現(xiàn)說這個數(shù)據(jù)就丟了。
所以此時一般是要求起碼設(shè)置如下 4 個參數(shù)：
給 topic 設(shè)置 replication.factor 參數(shù)：這個值必須大于 1，要求每個 partition 必須有至少 2 個副本。
在 Kafka 服務(wù)端設(shè)置 min.insync.replicas 參數(shù)：這個值必須大于 1，這個是要求一個 leader 至少感知到有至少一個 follower 還跟自己保持聯(lián)系，沒掉隊，這樣才能確保 leader 掛了還有一個 follower 吧。
在 producer 端設(shè)置 acks=all ：這個是要求每條數(shù)據(jù)，必須是寫入所有 replica 之后，才能認(rèn)為是寫成功了。
在 producer 端設(shè)置 retries=MAX （很大很大很大的一個值，無限次重試的意思）：這個是要求一旦寫入失敗，就無限重試，卡在這里了。
我們生產(chǎn)環(huán)境就是按照上述要求配置的，這樣配置之后，至少在 Kafka broker 端就可以保證在 leader 所在 broker 發(fā)生故障，進(jìn)行 leader 切換時，數(shù)據(jù)不會丟失。
2.1.3 消費(fèi)端弄丟了數(shù)據(jù)
唯一可能導(dǎo)致消費(fèi)者弄丟數(shù)據(jù)的情況，就是說，你消費(fèi)到了這個消息，然后消費(fèi)者那邊自動提交了 offset，讓 Kafka 以為你已經(jīng)消費(fèi)好了這個消息，但其實你才剛準(zhǔn)備處理這個消息，你還沒處理，你自己就掛了，此時這條消息就丟咯。
這不是跟 RabbitMQ 差不多嗎，大家都知道 Kafka 會自動提交 offset，那么只要關(guān)閉自動提交 offset，在處理完之后自己手動提交 offset，就可以保證數(shù)據(jù)不會丟。但是此時確實還是可能會有重復(fù)消費(fèi)，比如你剛處理完，還沒提交 offset，結(jié)果自己掛了，此時肯定會重復(fù)消費(fèi)一次，自己保證冪等性就好了。
生產(chǎn)環(huán)境碰到的一個問題，就是說我們的 Kafka 消費(fèi)者消費(fèi)到了數(shù)據(jù)之后是寫到一個內(nèi)存的 queue 里先緩沖一下，結(jié)果有的時候，你剛把消息寫入內(nèi)存 queue，然后消費(fèi)者會自動提交 offset。然后此時我們重啟了系統(tǒng)，就會導(dǎo)致內(nèi)存 queue 里還沒來得及處理的數(shù)據(jù)就丟失了。
文章來源：urlify.cn/yUnEVb
到此文章就結(jié)束了。如果今天的文章對你在進(jìn)階架構(gòu)師的路上有新的啟發(fā)和進(jìn)步，歡迎轉(zhuǎn)發(fā)給更多人。歡迎加入架構(gòu)師社區(qū)技術(shù)交流群，眾多大咖帶你進(jìn)階架構(gòu)師，在后臺回復(fù)“加群”即可入群。
這些年小編給你分享過的干貨
2.ERP系統(tǒng)，自帶進(jìn)銷存+財務(wù)+生產(chǎn)功能，拿來即用
3.帶工作流的SpringBoot后臺管理項目快速開發(fā)解決方案
4.最好的OA系統(tǒng)，拿來即用，非常方便
5.SpringBoot+Vue完整的外賣系統(tǒng)，手機(jī)端和后臺管理，附源碼！
轉(zhuǎn)發(fā)在看就是最大的支持??