欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

RocketMQ的push消費方式實現(xiàn)示例

 更新時間:2022年08月15日 16:14:41   作者:zzyang90  
這篇文章主要為大家介紹了RocketMQ的push消費方式實現(xiàn)示例詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步,早日升職加薪<BR>

引言

最近仍然暢游在RocketMQ的源碼中,這幾天剛好翻到了消費者的源碼,發(fā)現(xiàn)RocketMQ的對于push消費方式的實現(xiàn)簡直太聰明了,所以趁著我腦子里還有點印象的時候,趕緊來寫一篇文章,來掰扯一下,防止過兩天就忘得一干二凈了。

MQ消費方式

消費方式就是指消費者如何從MQ中獲取到消息,分為兩種方式,push(推方式)和pull(拉方式)。

1、push(推方式)

push,顧名思義,就是推的意思。就是當(dāng)MQ收到生產(chǎn)者產(chǎn)生的消息的時候,會主動將消息推送到消費者進行消費,這種模式就叫push,也就是MQ將消息推給到消費者的意思。

push模式

push這種模式的好處就是響應(yīng)快,消息的實時性比較高,一旦消息MQ收到消息,那么就能立馬將消息推送給消費者,消費者也就能立馬收到消息進行消費。

但是這種push的模式,有個缺點就是一旦消息量比較大時,對消費者性能要求比較高,因為是消費者無法控制MQ消息的推送速度,一旦消息量大,那么消費者消費消息的壓力就比較大。

2、pull(拉方式)

push是MQ主動給消費者推消息,那么pull呢?剛好跟push相反,就是消費者主動去MQ中拉取消息。

pull模式

那么pull的優(yōu)缺點自然也就跟push剛好相反。因為是消費者主動去MQ中拉取消息,那么消費者可根據(jù)自身消費的情況,決定何時去拉取消息,主動權(quán)在自己手上,這樣消費者的壓力就會相對小點;但是缺點也很明顯,那么就會實時性相對于push方式會低一些,因為你得決定拉的時間間隔。

其實想想,消費方式就跟拿快遞一樣,快遞就是一個消息,我自己就是消費者,快遞要么快遞小哥主動送(push)到家,要么我自己去快遞站拿(pull)。

RocketMQ對于消費方式的實現(xiàn)

上一節(jié)說了消費消息的兩種方式push和pull,或者說算一種理念。尚大的周陽老師有一句經(jīng)常說的話我比較贊同,那就是“天上飛的理念,必然有落地的實現(xiàn)”。所以push或者pull到底如何落地,得看具體的MQ的產(chǎn)品了。

而RocketMQ作為阿里開源的一款高性能、功能豐富的MQ,自然同時實現(xiàn)了push和pull的兩種消費方式,用戶可以選擇在項目中使用push還是pull。

push模式的實現(xiàn)

pull模式的實現(xiàn)

但是一般情況下,項目中都是使用push的方式來消費,因為pull除了時實性差外,pull方式還得讓開發(fā)人員主動去維護消息消費進度,增加額外的操作。

所以接下來就著重講一下RocketMQ是如何實現(xiàn)push的邏輯。

RocketMQ聰明地實現(xiàn)push的原因

上文說到push模式的優(yōu)點是時實性好,但是缺點就是消費者壓力會比較大,所以,難道實現(xiàn)push模式,只能舍棄壓力的控制么?

就在這時,RocketMQ大喊了一聲

是的,RocketMQ對于push模式做到了實時和壓力的平衡,這主要是因為RocketMQ的push模式其實算是一個“偽push”模式,真正底層的實現(xiàn)還是基于pull。

到這里可能有的小伙伴比較迷糊,怎么push變成“偽push”了,還是用pull實現(xiàn)的,到底是push還是pull?

前面我說過,push和pull只是一種理論,具體的實現(xiàn)看MQ。

所以RocketMQ為了兼顧兩者,就選擇通過消費者主動拉消息來實現(xiàn)push的效果,這也是為什么我稱為“偽push”的原因,RocketMQ都給封裝好了,讓你用起來感覺是MQ主動push消息給你的。

既然底層是pull,那么RokcetMQ在實現(xiàn)消費者的邏輯的時候,就可以很容易實現(xiàn)控制壓力的效果,畢竟這是“拉”方式天然自帶的buff;但是如何通過pull實現(xiàn)push的時實的優(yōu)點呢?畢竟魚和熊掌我RokcetMQ偏要兼得。

這時這就不得不提到一種叫“長輪詢”的機制。

輪詢與長輪詢

輪詢與長輪詢都屬于pull的實現(xiàn),都是由客戶端主動給服務(wù)端發(fā)送請求,拉取數(shù)據(jù)。套到MQ中,就是都是消費者主動去MQ拉消息。

輪詢

輪詢是指不管服務(wù)端數(shù)據(jù)有無更新,客戶端每隔定長時間請求拉取一次數(shù)據(jù),可能有更新數(shù)據(jù)返回,也可能什么都沒有。

再拿快遞舉例子,輪詢就好比,小明買的iphone 13 pro max快遞到了,顯示正在派送中,但是小明等不及了,于是就去快遞站拿,但是快遞還沒放到快遞站,但是小明的心里急啊,他忍受不了相思之苦,于是小明每隔5分鐘就往快遞站跑一次,問一下快遞到了沒,到了就拿回來。這就是輪詢的意思,也就是不論有沒有數(shù)據(jù),客戶端都會每隔一定時間去請求一次服務(wù)端。

來分析一下拿快遞的例子的問題:

  • 每隔5分鐘就往快遞站跑,那不是累死個小明么。
  • 還有一個問題,假設(shè)剛跑到快遞站,快遞沒到,就回去了,但是剛到家的時候,快遞到了,于是又等了5分鐘,再去快遞站終于拿到快遞了,但是其實快遞都到了幾分鐘了,你還是沒有第一時間拿到快遞,這就造成了延遲。

從而對應(yīng)到程序中,就是會產(chǎn)生如下問題

  • 對于消息而言,會一直產(chǎn)生,這就要求消費者不停地間隔一定時間去拉取消息,即使沒有消息也需要去請求,就會造成大量無用的請求,白白浪費大量耗費服務(wù)器內(nèi)存和寬帶資源。
  • 可能造成數(shù)據(jù)的延遲

長輪詢

說長輪詢概念之前,先來救救小明吧,畢竟小明可不想狗帶。

既然原先小明每隔5分鐘跑一次,那么是不是可以換種思路,當(dāng)快遞還沒到的時候,讓小明不要回來,直接在快遞站待著,當(dāng)快遞到的時候,才讓小明拿著快遞回家。這下小明就喜死了,既可以有時間刷刷某音,逛逛某東,還可以在第一時間拿到13 pro max。

所以這種可以在快遞站等待的機制,就叫長輪詢。

長輪詢也是客戶端請求服務(wù)端,如果服務(wù)端有數(shù)據(jù),那么就立馬返回,客戶端再次請求;當(dāng)服務(wù)端不存在數(shù)據(jù)的時候,服務(wù)端并不會給客戶端響應(yīng),而是將請求給hold住,當(dāng)服務(wù)端有數(shù)據(jù)的時候才會給客戶端響應(yīng),返回數(shù)據(jù)。

所以長輪詢可以解決如下問題

  • 解決輪詢帶來的頻繁請求服務(wù)端但是沒有的問題
  • 一旦新的數(shù)據(jù)到了,那么消費者能立馬就可以獲取到新的數(shù)據(jù),所以從效果上,有點像是push的感覺。

但是長輪詢也會帶來服務(wù)端代碼實現(xiàn)邏輯復(fù)雜的問題,當(dāng)然相比于優(yōu)點來說,都不太重要。

push消費方式源碼探究

理論都講完了,接下來就到了show me the code的時間了,來看看RocketMQ的是如何通過長輪詢機制來實現(xiàn)壓力和時實的平衡。

這里我畫了一張push模式下消費者消費流程圖。

消費者拉取消息的邏輯

  • ①消費者有一個后臺線程,會去處理拉取消息(PullRequest)
  • ②先去判斷有沒有過多消息沒有消費,如果有的話,那么就間隔一定時間再次從①開始執(zhí)行拉取消息的邏輯
  • ③消費者沒有過多消息沒有消費,那么就會直接向MQ發(fā)送拉取消息的請求,有消息就返回,沒有消息就hold住請求,等有新的消息到的時候才返回
  • ④消費者獲取到消息之后,會去找用戶自定義的消息處理邏輯的實現(xiàn)(MessageListener的實現(xiàn))去消費消息,同時會再次拉取消息,繼續(xù)從①開始執(zhí)行邏輯

1、消費者拉取消息控制壓力源碼

當(dāng)消費者準(zhǔn)備去拉消息的時候,會先去判斷當(dāng)前消費者消費的壓力再決定是否去拉取消息。

RocketMQ提供了兩種判斷消費壓力邏輯,一種是基于還未消費的消息的數(shù)量的大小,還有一種是基于還未消費的消息所占內(nèi)存的大小。

控制壓力源碼

  • 判斷還未消費消息的數(shù)量,數(shù)量太多就等會再執(zhí)行重新執(zhí)行拉取消息的邏輯
  • 判斷還未消費消息的大小,如果還未消息的消息占用的內(nèi)存過大,就等會再執(zhí)行重新執(zhí)行拉取消息的邏輯

總的一句話就是,當(dāng)消費者消費的壓力過大時,就不會去拉取消息,而是等待一定的時間再去執(zhí)行拉取消息的邏輯,如果壓力還是很大,就還繼續(xù)等,如此循環(huán),直到消費者的消費壓力小于閾值的時候,才會真正的發(fā)送請求到MQ中拉取消息。

2、MQ將請求hold住源碼

當(dāng)服務(wù)端未找到消息時,就將請求進行掛起,存起來

請求hold住源碼

拉取不到消息時,會調(diào)用PullRequestHoldService的suspendPullRequest方法講請求存儲起來。PullRequestHoldService是用來存儲拉取請求的類。

PullRequestHoldService

suspendPullRequest方法會將請求分類,放到ManyPullRequest里,然后用一個ConcurrentHashMap進行存儲

3、MQ收到消息響應(yīng)給消費者的源碼

NotifyMessageArrivingListener

當(dāng)生產(chǎn)者發(fā)送的消息達(dá)到MQ的時候,MQ會回調(diào)NotifyMessageArrivingListener的arriving方法,之后就會調(diào)用PullRequestHoldService的notifyMessageArriving方法,MQ會重新處理拉取消息的邏輯,此時就能找到最新來的那條消息,從而將最新的消息通過網(wǎng)絡(luò)返回給消費者。

notifyMessageArriving和返回消息邏輯

最后

所以從以上的分析可以看出,RocketMQ對于push的消費方式的實現(xiàn)是基于長輪詢機制來實現(xiàn)的,同時平衡了時實和壓力,這其實就很nice了。

最后我想說一句,其實不論是pull還是push,又或是輪詢和長輪詢,其實都是一種理論或者說是一種思想,不單單是MQ的東西,就比如在Nacos中,也使用了push和長輪詢機制。但是這些理論在不同產(chǎn)品的具體實現(xiàn),實現(xiàn)方式可能不太一樣,但都是大同小異,所以當(dāng)你懂了這些思想,再看其它框架的源碼,其實就很容易了。

以上就是RocketMQ的push消費方式實現(xiàn)示例的詳細(xì)內(nèi)容,更多關(guān)于RocketMQ push消費方式的資料請關(guān)注腳本之家其它相關(guān)文章!

相關(guān)文章

最新評論