RocketMQ的push消費方式如何實現

本篇內容介紹了“RocketMQ的push消費方式如何實現”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學有所成！

MQ消費方式

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

1、push（推方式）

push，顧名思義，就是推的意思。就是當MQ收到生產者產生的消息的時候，會主動將消息推送到消費者進行消費，這種模式就叫push，也就是MQ將消息推給到消費者的意思。

push模式

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

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

2、pull（拉方式）

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

pull模式

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

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

RocketMQ對于消費方式的實現

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

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

push模式的實現

pull模式的實現

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

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

RocketMQ聰明地實現push的原因

上文說到push模式的優點是時實性好，但是缺點就是消費者壓力會比較大，所以，難道實現push模式，只能舍棄壓力的控制么？

就在這時，RocketMQ大喊了一聲

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

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

前面我說過，push和pull只是一種理論，具體的實現看MQ。

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

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

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

輪詢與長輪詢

輪詢與長輪詢都屬于pull的實現，都是由客戶端主動給服務端發送請求，拉取數據。套到MQ中，就是都是消費者主動去MQ拉消息。

輪詢

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

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

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

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

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

從而對應到程序中，就是會產生如下問題

• 對于消息而言，會一直產生，這就要求消費者不停地間隔一定時間去拉取消息，即使沒有消息也需要去請求，就會造成大量無用的請求，白白浪費大量耗費服務器內存和寬帶資源。

• 可能造成數據的延遲

長輪詢

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

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

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

長輪詢也是客戶端請求服務端，如果服務端有數據，那么就立馬返回，客戶端再次請求；當服務端不存在數據的時候，服務端并不會給客戶端響應，而是將請求給hold住，當服務端有數據的時候才會給客戶端響應，返回數據。

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

• 解決輪詢帶來的頻繁請求服務端但是沒有的問題

• 一旦新的數據到了，那么消費者能立馬就可以獲取到新的數據，所以從效果上，有點像是push的感覺。

但是長輪詢也會帶來服務端代碼實現邏輯復雜的問題，當然相比于優點來說，都不太重要。

push消費方式源碼探究

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

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

消費者拉取消息的邏輯

• ①消費者有一個后臺線程，會去處理拉取消息（PullRequest）

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

• ③消費者沒有過多消息沒有消費，那么就會直接向MQ發送拉取消息的請求，有消息就返回，沒有消息就hold住請求，等有新的消息到的時候才返回

• ④消費者獲取到消息之后，會去找用戶自定義的消息處理邏輯的實現（MessageListener的實現）去消費消息，同時會再次拉取消息，繼續從①開始執行邏輯

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

當消費者準備去拉消息的時候，會先去判斷當前消費者消費的壓力再決定是否去拉取消息。

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

控制壓力源碼

• 判斷還未消費消息的數量，數量太多就等會再執行重新執行拉取消息的邏輯

• 判斷還未消費消息的大小，如果還未消息的消息占用的內存過大，就等會再執行重新執行拉取消息的邏輯

總的一句話就是，當消費者消費的壓力過大時，就不會去拉取消息，而是等待一定的時間再去執行拉取消息的邏輯，如果壓力還是很大，就還繼續等，如此循環，直到消費者的消費壓力小于閾值的時候，才會真正的發送請求到MQ中拉取消息。

2、MQ將請求hold住源碼

當服務端未找到消息時，就將請求進行掛起，存起來

請求hold住源碼

拉取不到消息時，會調用PullRequestHoldService的suspendPullRequest方法講請求存儲起來。PullRequestHoldService是用來存儲拉取請求的類。

PullRequestHoldService

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

3、MQ收到消息響應給消費者的源碼

NotifyMessageArrivingListener

當生產者發送的消息達到MQ的時候，MQ會回調NotifyMessageArrivingListener的arriving方法，之后就會調用PullRequestHoldService的notifyMessageArriving方法，MQ會重新處理拉取消息的邏輯，此時就能找到最新來的那條消息，從而將最新的消息通過網絡返回給消費者。

notifyMessageArriving和返回消息邏輯

“RocketMQ的push消費方式如何實現”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識可以關注億速云網站，小編將為大家輸出更多高質量的實用文章！