RocketMQ的push消費(fèi)方式如何實(shí)現(xiàn)

本篇內(nèi)容介紹了“RocketMQ的push消費(fèi)方式如何實(shí)現(xiàn)”的有關(guān)知識(shí)，在實(shí)際案例的操作過程中，不少人都會(huì)遇到這樣的困境，接下來(lái)就讓小編帶領(lǐng)大家學(xué)習(xí)一下如何處理這些情況吧！希望大家仔細(xì)閱讀，能夠?qū)W有所成！

MQ消費(fèi)方式

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

1、push（推方式）

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

push模式

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

但是這種push的模式，有個(gè)缺點(diǎn)就是一旦消息量比較大時(shí)，對(duì)消費(fèi)者性能要求比較高，因?yàn)槭窍M(fèi)者無(wú)法控制MQ消息的推送速度，一旦消息量大，那么消費(fèi)者消費(fèi)消息的壓力就比較大。

2、pull（拉方式）

push是MQ主動(dòng)給消費(fèi)者推消息，那么pull呢？剛好跟push相反，就是消費(fèi)者主動(dòng)去MQ中拉取消息。

pull模式

那么pull的優(yōu)缺點(diǎn)自然也就跟push剛好相反。因?yàn)槭窍M(fèi)者主動(dòng)去MQ中拉取消息，那么消費(fèi)者可根據(jù)自身消費(fèi)的情況，決定何時(shí)去拉取消息，主動(dòng)權(quán)在自己手上，這樣消費(fèi)者的壓力就會(huì)相對(duì)小點(diǎn)；但是缺點(diǎn)也很明顯，那么就會(huì)實(shí)時(shí)性相對(duì)于push方式會(huì)低一些，因?yàn)槟愕脹Q定拉的時(shí)間間隔。

其實(shí)想想，消費(fèi)方式就跟拿快遞一樣，快遞就是一個(gè)消息，我自己就是消費(fèi)者，快遞要么快遞小哥主動(dòng)送（push）到家，要么我自己去快遞站拿（pull）。

RocketMQ對(duì)于消費(fèi)方式的實(shí)現(xiàn)

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

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

push模式的實(shí)現(xiàn)

pull模式的實(shí)現(xiàn)

但是一般情況下，項(xiàng)目中都是使用push的方式來(lái)消費(fèi)，因?yàn)閜ull除了時(shí)實(shí)性差外，pull方式還得讓開發(fā)人員主動(dòng)去維護(hù)消息消費(fèi)進(jìn)度，增加額外的操作。

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

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

上文說(shuō)到push模式的優(yōu)點(diǎn)是時(shí)實(shí)性好，但是缺點(diǎn)就是消費(fèi)者壓力會(huì)比較大，所以，難道實(shí)現(xiàn)push模式，只能舍棄壓力的控制么？

就在這時(shí)，RocketMQ大喊了一聲

是的，RocketMQ對(duì)于push模式做到了實(shí)時(shí)和壓力的平衡，這主要是因?yàn)镽ocketMQ的push模式其實(shí)算是一個(gè)“偽push”模式，真正底層的實(shí)現(xiàn)還是基于pull。

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

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

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

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

這時(shí)這就不得不提到一種叫“長(zhǎng)輪詢”的機(jī)制。

輪詢與長(zhǎng)輪詢

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

輪詢

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

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

來(lái)分析一下拿快遞的例子的問題：

• 每隔5分鐘就往快遞站跑，那不是累死個(gè)小明么。

• 還有一個(gè)問題，假設(shè)剛跑到快遞站，快遞沒到，就回去了，但是剛到家的時(shí)候，快遞到了，于是又等了5分鐘，再去快遞站終于拿到快遞了，但是其實(shí)快遞都到了幾分鐘了，你還是沒有第一時(shí)間拿到快遞，這就造成了延遲。

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

• 對(duì)于消息而言，會(huì)一直產(chǎn)生，這就要求消費(fèi)者不停地間隔一定時(shí)間去拉取消息，即使沒有消息也需要去請(qǐng)求，就會(huì)造成大量無(wú)用的請(qǐng)求，白白浪費(fèi)大量耗費(fèi)服務(wù)器內(nèi)存和寬帶資源。

• 可能造成數(shù)據(jù)的延遲

長(zhǎng)輪詢

說(shuō)長(zhǎng)輪詢概念之前，先來(lái)救救小明吧，畢竟小明可不想狗帶。

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

所以這種可以在快遞站等待的機(jī)制，就叫長(zhǎng)輪詢。

長(zhǎng)輪詢也是客戶端請(qǐng)求服務(wù)端，如果服務(wù)端有數(shù)據(jù)，那么就立馬返回，客戶端再次請(qǐng)求；當(dāng)服務(wù)端不存在數(shù)據(jù)的時(shí)候，服務(wù)端并不會(huì)給客戶端響應(yīng)，而是將請(qǐng)求給hold住，當(dāng)服務(wù)端有數(shù)據(jù)的時(shí)候才會(huì)給客戶端響應(yīng)，返回?cái)?shù)據(jù)。

所以長(zhǎng)輪詢可以解決如下問題

• 解決輪詢帶來(lái)的頻繁請(qǐng)求服務(wù)端但是沒有的問題

• 一旦新的數(shù)據(jù)到了，那么消費(fèi)者能立馬就可以獲取到新的數(shù)據(jù)，所以從效果上，有點(diǎn)像是push的感覺。

但是長(zhǎng)輪詢也會(huì)帶來(lái)服務(wù)端代碼實(shí)現(xiàn)邏輯復(fù)雜的問題，當(dāng)然相比于優(yōu)點(diǎn)來(lái)說(shuō)，都不太重要。

push消費(fèi)方式源碼探究

理論都講完了，接下來(lái)就到了show me the code的時(shí)間了，來(lái)看看RocketMQ的是如何通過長(zhǎng)輪詢機(jī)制來(lái)實(shí)現(xiàn)壓力和時(shí)實(shí)的平衡。

這里我畫了一張push模式下消費(fèi)者消費(fèi)流程圖。

消費(fèi)者拉取消息的邏輯

• ①消費(fèi)者有一個(gè)后臺(tái)線程，會(huì)去處理拉取消息（PullRequest）

• ②先去判斷有沒有過多消息沒有消費(fèi)，如果有的話，那么就間隔一定時(shí)間再次從①開始執(zhí)行拉取消息的邏輯

• ③消費(fèi)者沒有過多消息沒有消費(fèi)，那么就會(huì)直接向MQ發(fā)送拉取消息的請(qǐng)求，有消息就返回，沒有消息就hold住請(qǐng)求，等有新的消息到的時(shí)候才返回

• ④消費(fèi)者獲取到消息之后，會(huì)去找用戶自定義的消息處理邏輯的實(shí)現(xiàn)（MessageListener的實(shí)現(xiàn)）去消費(fèi)消息，同時(shí)會(huì)再次拉取消息，繼續(xù)從①開始執(zhí)行邏輯

1、消費(fèi)者拉取消息控制壓力源碼

當(dāng)消費(fèi)者準(zhǔn)備去拉消息的時(shí)候，會(huì)先去判斷當(dāng)前消費(fèi)者消費(fèi)的壓力再?zèng)Q定是否去拉取消息。

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

控制壓力源碼

• 判斷還未消費(fèi)消息的數(shù)量，數(shù)量太多就等會(huì)再執(zhí)行重新執(zhí)行拉取消息的邏輯

• 判斷還未消費(fèi)消息的大小，如果還未消息的消息占用的內(nèi)存過大，就等會(huì)再執(zhí)行重新執(zhí)行拉取消息的邏輯

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

2、MQ將請(qǐng)求hold住源碼

當(dāng)服務(wù)端未找到消息時(shí)，就將請(qǐng)求進(jìn)行掛起，存起來(lái)

請(qǐng)求hold住源碼

拉取不到消息時(shí)，會(huì)調(diào)用PullRequestHoldService的suspendPullRequest方法講請(qǐng)求存儲(chǔ)起來(lái)。PullRequestHoldService是用來(lái)存儲(chǔ)拉取請(qǐng)求的類。

PullRequestHoldService

suspendPullRequest方法會(huì)將請(qǐng)求分類，放到ManyPullRequest里，然后用一個(gè)ConcurrentHashMap進(jìn)行存儲(chǔ)

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

NotifyMessageArrivingListener

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

notifyMessageArriving和返回消息邏輯

“RocketMQ的push消費(fèi)方式如何實(shí)現(xiàn)”的內(nèi)容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識(shí)可以關(guān)注億速云網(wǎng)站，小編將為大家輸出更多高質(zhì)量的實(shí)用文章！