消息隊列中如何保證消息不丟失?

rabbitmq

1)生產者丟了數據

生產者將數據發送到rabbitmq的時候,可能數據就在半路給搞丟了,因為網絡啥的問題,都有可能。

此時可以選擇用rabbitmq提供的事務功能,就是生產者發送數據之前開啟rabbitmq事務(channel.txSelect),然後發送消息,如果消息沒有成功被rabbitmq接收到,那麼生產者會收到異常報錯,此時就可以回滾事務(channel.txRollback),然後重試發送消息;如果收到了消息,那麼可以提交事務(channel.txCommit)。但是問題是,rabbitmq事務機制一搞,基本上吞吐量會下來,因為太耗性能。

所以一般來說,如果你要確保說寫rabbitmq的消息別丟,可以開啟confirm模式,在生產者那裡設置開啟confirm模式之後,你每次寫的消息都會分配一個唯一的id,然後如果寫入了rabbitmq中,rabbitmq會給你回傳一個ack消息,告訴你說這個消息ok了。如果rabbitmq沒能處理這個消息,會回調你一個nack接口,告訴你這個消息接收失敗,你可以重試。而且你可以結合這個機制自己在內存裡維護每個消息id的狀態,如果超過一定時間還沒接收到這個消息的回調,那麼你可以重發。

事務機制和cnofirm機制最大的不同在於,事務機制是同步的,你提交一個事務之後會阻塞在那兒,但是confirm機制是異步的,你發送個消息之後就可以發送下一個消息,然後那個消息rabbitmq接收了之後會異步回調你一個接口通知你這個消息接收到了。

所以一般在生產者這塊避免數據丟失,都是用confirm機制的。

2)rabbitmq弄丟了數據

就是rabbitmq自己弄丟了數據,這個你必須開啟rabbitmq的持久化,就是消息寫入之後會持久化到磁盤,哪怕是rabbitmq自己掛了,恢復之後會自動讀取之前存儲的數據,一般數據不會丟。除非極其罕見的是,rabbitmq還沒持久化,自己就掛了,可能導致少量數據會丟失的,但是這個概率較小。

設置持久化有兩個步驟,第一個是創建queue的時候將其設置為持久化的,這樣就可以保證rabbitmq持久化queue的元數據,但是不會持久化queue裡的數據;第二個是發送消息的時候將消息的deliveryMode設置為2,就是將消息設置為持久化的,此時rabbitmq就會將消息持久化到磁盤上去。必須要同時設置這兩個持久化才行,rabbitmq哪怕是掛了,再次重啟,也會從磁盤上重啟恢復queue,恢復這個queue裡的數據。

而且持久化可以跟生產者那邊的confirm機制配合起來,只有消息被持久化到磁盤之後,才會通知生產者ack了,所以哪怕是在持久化到磁盤之前,rabbitmq掛了,數據丟了,生產者收不到ack,你也是可以自己重發的。

哪怕是你給rabbitmq開啟了持久化機制,也有一種可能,就是這個消息寫到了rabbitmq中,但是還沒來得及持久化到磁盤上,結果不巧,此時rabbitmq掛了,就會導致內存裡的一點點數據會丟失。

3)消費端弄丟了數據

rabbitmq如果丟失了數據,主要是因為你消費的時候,剛消費到,還沒處理,結果進程掛了,比如重啟了,那麼就尷尬了,rabbitmq認為你都消費了,這數據就丟了。

這個時候得用rabbitmq提供的ack機制,簡單來說,就是你關閉rabbitmq自動ack,可以通過一個api來調用就行,然後每次你自己代碼裡確保處理完的時候,再程序裡ack一把。這樣的話,如果你還沒處理完,不就沒有ack?那rabbitmq就認為你還沒處理完,這個時候rabbitmq會把這個消費分配給別的consumer去處理,消息是不會丟的。

kafka

1)消費端弄丟了數據

唯一可能導致消費者弄丟數據的情況,就是說,你那個消費到了這個消息,然後消費者那邊自動提交了offset,讓kafka以為你已經消費好了這個消息,其實你剛準備處理這個消息,你還沒處理,你自己就掛了,此時這條消息就丟咯。

這不是一樣麼,大家都知道kafka會自動提交offset,那麼只要關閉自動提交offset,在處理完之後自己手動提交offset,就可以保證數據不會丟。但是此時確實還是會重複消費,比如你剛處理完,還沒提交offset,結果自己掛了,此時肯定會重複消費一次,自己保證冪等性就好了。

生產環境碰到的一個問題,就是說我們的kafka消費者消費到了數據之後是寫到一個內存的queue裡先緩衝一下,結果有的時候,你剛把消息寫入內存queue,然後消費者會自動提交offset。

然後此時我們重啟了系統,就會導致內存queue裡還沒來得及處理的數據就丟失了

2)kafka弄丟了數據

這塊比較常見的一個場景,就是kafka某個broker宕機,然後重新選舉partiton的leader時。大家想想,要是此時其他的follower剛好還有些數據沒有同步,結果此時leader掛了,然後選舉某個follower成leader之後,他不就少了一些數據?這就丟了一些數據啊。

生產環境也遇到過,我們也是,之前kafka的leader機器宕機了,將follower切換為leader之後,就會發現說這個數據就丟了

所以此時一般是要求起碼設置如下4個參數:

給這個topic設置replication.factor參數:這個值必須大於1,要求每個partition必須有至少2個副本

在kafka服務端設置min.insync.replicas參數:這個值必須大於1,這個是要求一個leader至少感知到有至少一個follower還跟自己保持聯繫,沒掉隊,這樣才能確保leader掛了還有一個follower吧

在producer端設置acks=all:這個是要求每條數據,必須是寫入所有replica之後,才能認為是寫成功了

在producer端設置retries=MAX(很大很大很大的一個值,無限次重試的意思):這個是要求一旦寫入失敗,就無限重試,卡在這裡了

我們生產環境就是按照上述要求配置的,這樣配置之後,至少在kafka broker端就可以保證在leader所在broker發生故障,進行leader切換時,數據不會丟失

3)生產者會不會弄丟數據

如果按照上述的思路設置了ack=all,一定不會丟,要求是,你的leader接收到消息,所有的follower都同步到了消息之後,才認為本次寫成功了。如果沒滿足這個條件,生產者會自動不斷的重試,重試無限次。


分享到:


相關文章: