本篇內(nèi)容主要講解“RabbitMQ和Kafka怎么保證消息隊(duì)列的可靠性傳輸”,感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷,實(shí)用性強(qiáng)。下面就讓小編來帶大家學(xué)習(xí)“RabbitMQ和Kafka怎么保證消息隊(duì)列的可靠性傳輸”吧!
創(chuàng)新互聯(lián)建站服務(wù)項(xiàng)目包括高邑網(wǎng)站建設(shè)、高邑網(wǎng)站制作、高邑網(wǎng)頁制作以及高邑網(wǎng)絡(luò)營銷策劃等。多年來,我們專注于互聯(lián)網(wǎng)行業(yè),利用自身積累的技術(shù)優(yōu)勢、行業(yè)經(jīng)驗(yàn)、深度合作伙伴關(guān)系等,向廣大中小型企業(yè)、政府機(jī)構(gòu)等提供互聯(lián)網(wǎng)行業(yè)的解決方案,高邑網(wǎng)站推廣取得了明顯的社會效益與經(jīng)濟(jì)效益。目前,我們服務(wù)的客戶以成都為中心已經(jīng)輻射到高邑省份的部分城市,未來相信會繼續(xù)擴(kuò)大服務(wù)區(qū)域并繼續(xù)獲得客戶的支持與信任!如何保證消息的可靠性傳輸?或者說,如何處理消息丟失的問題?
這個(gè)是肯定的,用 MQ 有個(gè)基本原則,就是數(shù)據(jù)不能多一條,也不能少一條,不能多,就是前面說的重復(fù)消費(fèi)和冪等性問題。不能少,就是說這數(shù)據(jù)別搞丟了。那這個(gè)問題你必須得考慮一下。
如果說你這個(gè)是用 MQ 來傳遞非常核心的消息,比如說計(jì)費(fèi)、扣費(fèi)的一些消息,那必須確保這個(gè) MQ 傳遞過程中絕對不會把計(jì)費(fèi)消息給弄丟。
數(shù)據(jù)的丟失問題,可能出現(xiàn)在生產(chǎn)者、MQ、消費(fèi)者中,咱們從 RabbitMQ 和 Kafka 分別來分析一下吧。
生產(chǎn)者將數(shù)據(jù)發(fā)送到 RabbitMQ 的時(shí)候,可能數(shù)據(jù)就在半路給搞丟了,因?yàn)榫W(wǎng)絡(luò)問題啥的,都有可能。
此時(shí)可以選擇用 RabbitMQ 提供的事務(wù)功能,就是生產(chǎn)者發(fā)送數(shù)據(jù)之前開啟 RabbitMQ 事務(wù)channel.txSelect
,然后發(fā)送消息,如果消息沒有成功被 RabbitMQ 接收到,那么生產(chǎn)者會收到異常報(bào)錯(cuò),此時(shí)就可以回滾事務(wù)channel.txRollback
,然后重試發(fā)送消息;如果收到了消息,那么可以提交事務(wù)channel.txCommit
。
// 開啟事務(wù) channel.txSelect try { // 這里發(fā)送消息 } catch (Exception e) { channel.txRollback // 這里再次重發(fā)這條消息 } // 提交事務(wù) channel.txCommit
但是問題是,RabbitMQ 事務(wù)機(jī)制(同步)一搞,基本上吞吐量會下來,因?yàn)樘男阅堋?/p>
所以一般來說,如果你要確保說寫 RabbitMQ 的消息別丟,可以開啟 confirm
模式,在生產(chǎn)者那里設(shè)置開啟 confirm
模式之后,你每次寫的消息都會分配一個(gè)唯一的 id,然后如果寫入了 RabbitMQ 中,RabbitMQ 會給你回傳一個(gè) ack
消息,告訴你說這個(gè)消息 ok 了。如果 RabbitMQ 沒能處理這個(gè)消息,會回調(diào)你的一個(gè) nack
接口,告訴你這個(gè)消息接收失敗,你可以重試。而且你可以結(jié)合這個(gè)機(jī)制自己在內(nèi)存里維護(hù)每個(gè)消息 id 的狀態(tài),如果超過一定時(shí)間還沒接收到這個(gè)消息的回調(diào),那么你可以重發(fā)。
事務(wù)機(jī)制和 cnofirm
機(jī)制大的不同在于,事務(wù)機(jī)制是同步的,你提交一個(gè)事務(wù)之后會阻塞在那兒,但是 confirm
機(jī)制是異步的,你發(fā)送個(gè)消息之后就可以發(fā)送下一個(gè)消息,然后那個(gè)消息 RabbitMQ 接收了之后會異步回調(diào)你的一個(gè)接口通知你這個(gè)消息接收到了。
所以一般在生產(chǎn)者這塊避免數(shù)據(jù)丟失,都是用 confirm
機(jī)制的。
就是 RabbitMQ 自己弄丟了數(shù)據(jù),這個(gè)你必須開啟 RabbitMQ 的持久化,就是消息寫入之后會持久化到磁盤,哪怕是 RabbitMQ 自己掛了,恢復(fù)之后會自動(dòng)讀取之前存儲的數(shù)據(jù),一般數(shù)據(jù)不會丟。除非極其罕見的是,RabbitMQ 還沒持久化,自己就掛了,可能導(dǎo)致少量數(shù)據(jù)丟失,但是這個(gè)概率較小。
設(shè)置持久化有兩個(gè)步驟:
創(chuàng)建 queue 的時(shí)候?qū)⑵湓O(shè)置為持久化
這樣就可以保證 RabbitMQ 持久化 queue 的元數(shù)據(jù),但是它是不會持久化 queue 里的數(shù)據(jù)的。
第二個(gè)是發(fā)送消息的時(shí)候?qū)⑾⒌?nbsp;deliveryMode
設(shè)置為 2
就是將消息設(shè)置為持久化的,此時(shí) RabbitMQ 就會將消息持久化到磁盤上去。
必須要同時(shí)設(shè)置這兩個(gè)持久化才行,RabbitMQ 哪怕是掛了,再次重啟,也會從磁盤上重啟恢復(fù) queue,恢復(fù)這個(gè) queue 里的數(shù)據(jù)。
注意,哪怕是你給 RabbitMQ 開啟了持久化機(jī)制,也有一種可能,就是這個(gè)消息寫到了 RabbitMQ 中,但是還沒來得及持久化到磁盤上,結(jié)果不巧,此時(shí) RabbitMQ 掛了,就會導(dǎo)致內(nèi)存里的一點(diǎn)點(diǎn)數(shù)據(jù)丟失。
所以,持久化可以跟生產(chǎn)者那邊的 confirm
機(jī)制配合起來,只有消息被持久化到磁盤之后,才會通知生產(chǎn)者 ack
了,所以哪怕是在持久化到磁盤之前,RabbitMQ 掛了,數(shù)據(jù)丟了,生產(chǎn)者收不到 ack
,你也是可以自己重發(fā)的。
RabbitMQ 如果丟失了數(shù)據(jù),主要是因?yàn)槟阆M(fèi)的時(shí)候,剛消費(fèi)到,還沒處理,結(jié)果進(jìn)程掛了,比如重啟了,那么就尷尬了,RabbitMQ 認(rèn)為你都消費(fèi)了,這數(shù)據(jù)就丟了。
這個(gè)時(shí)候得用 RabbitMQ 提供的 ack
機(jī)制,簡單來說,就是你必須關(guān)閉 RabbitMQ 的自動(dòng) ack
,可以通過一個(gè) api 來調(diào)用就行,然后每次你自己代碼里確保處理完的時(shí)候,再在程序里 ack
一把。這樣的話,如果你還沒處理完,不就沒有 ack
了?那 RabbitMQ 就認(rèn)為你還沒處理完,這個(gè)時(shí)候 RabbitMQ 會把這個(gè)消費(fèi)分配給別的 consumer 去處理,消息是不會丟的。
唯一可能導(dǎo)致消費(fèi)者弄丟數(shù)據(jù)的情況,就是說,你消費(fèi)到了這個(gè)消息,然后消費(fèi)者那邊自動(dòng)提交了 offset,讓 Kafka 以為你已經(jīng)消費(fèi)好了這個(gè)消息,但其實(shí)你才剛準(zhǔn)備處理這個(gè)消息,你還沒處理,你自己就掛了,此時(shí)這條消息就丟咯。
這不是跟 RabbitMQ 差不多嗎,大家都知道 Kafka 會自動(dòng)提交 offset,那么只要關(guān)閉自動(dòng)提交offset,在處理完之后自己手動(dòng)提交 offset,就可以保證數(shù)據(jù)不會丟。但是此時(shí)確實(shí)還是可能會有重復(fù)消費(fèi),比如你剛處理完,還沒提交 offset,結(jié)果自己掛了,此時(shí)肯定會重復(fù)消費(fèi)一次,自己保證冪等性就好了。
生產(chǎn)環(huán)境碰到的一個(gè)問題,就是說我們的 Kafka 消費(fèi)者消費(fèi)到了數(shù)據(jù)之后是寫到一個(gè)內(nèi)存的 queue 里先緩沖一下,結(jié)果有的時(shí)候,你剛把消息寫入內(nèi)存 queue,然后消費(fèi)者會自動(dòng)提交 offset。然后此時(shí)我們重啟了系統(tǒng),就會導(dǎo)致內(nèi)存 queue 里還沒來得及處理的數(shù)據(jù)就丟失了。
這塊比較常見的一個(gè)場景,就是 Kafka 某個(gè) broker 宕機(jī),然后重新選舉 partition 的 leader。大家想想,要是此時(shí)其他的 follower 剛好還有些數(shù)據(jù)沒有同步,結(jié)果此時(shí) leader 掛了,然后選舉某個(gè) follower 成 leader 之后,不就少了一些數(shù)據(jù)?這就丟了一些數(shù)據(jù)啊。
生產(chǎn)環(huán)境也遇到過,我們也是,之前 Kafka 的 leader 機(jī)器宕機(jī)了,將 follower 切換為 leader 之后,就會發(fā)現(xiàn)說這個(gè)數(shù)據(jù)就丟了。
所以此時(shí)一般是要求起碼設(shè)置如下 4 個(gè)參數(shù):
給 topic 設(shè)置 replication.factor
參數(shù):這個(gè)值必須大于 1,要求每個(gè) partition 必須有至少 2 個(gè)副本。
在 Kafka 服務(wù)端設(shè)置 min.insync.replicas
參數(shù):這個(gè)值必須大于 1,這個(gè)是要求一個(gè) leader 至少感知到有至少一個(gè) follower 還跟自己保持聯(lián)系,沒掉隊(duì),這樣才能確保 leader 掛了還有一個(gè) follower 吧。
在 producer 端設(shè)置 acks=all
:這個(gè)是要求每條數(shù)據(jù),必須是寫入所有 replica 之后,才能認(rèn)為是寫成功了。
在 producer 端設(shè)置 retries=MAX
(很大很大很大的一個(gè)值,無限次重試的意思):這個(gè)是要求一旦寫入失敗,就無限重試,卡在這里了。
我們生產(chǎn)環(huán)境就是按照上述要求配置的,這樣配置之后,至少在 Kafka broker 端就可以保證在 leader 所在 broker 發(fā)生故障,進(jìn)行 leader 切換時(shí),數(shù)據(jù)不會丟失。
如果按照上述的思路設(shè)置了 acks=all
,一定不會丟,要求是,你的 leader 接收到消息,所有的 follower 都同步到了消息之后,才認(rèn)為本次寫成功了。如果沒滿足這個(gè)條件,生產(chǎn)者會自動(dòng)不斷的重試,重試無限次。
到此,相信大家對“RabbitMQ和Kafka怎么保證消息隊(duì)列的可靠性傳輸”有了更深的了解,不妨來實(shí)際操作一番吧!這里是創(chuàng)新互聯(lián)網(wǎng)站,更多相關(guān)內(nèi)容可以進(jìn)入相關(guān)頻道進(jìn)行查詢,關(guān)注我們,繼續(xù)學(xué)習(xí)!
當(dāng)前標(biāo)題:RabbitMQ和Kafka怎么保證消息隊(duì)列的可靠性傳輸-創(chuàng)新互聯(lián)
網(wǎng)頁地址:http://www.rwnh.cn/article20/ceigjo.html
成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián),為您提供電子商務(wù)、ChatGPT、云服務(wù)器、App設(shè)計(jì)、做網(wǎng)站、定制開發(fā)
聲明:本網(wǎng)站發(fā)布的內(nèi)容(圖片、視頻和文字)以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主,如果涉及侵權(quán)請盡快告知,我們將會在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場,如需處理請聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載,或轉(zhuǎn)載時(shí)需注明來源: 創(chuàng)新互聯(lián)
猜你還喜歡下面的內(nèi)容