RocketMQ的push消費方式實現的太聰明瞭
大家好，我是三友，我又來了~~
最近仍然暢遊在RocketMQ的原始碼中，這幾天剛好翻到了消費者的原始碼，發現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的消費方式的實現是基於長輪詢機制來實現的，同時平衡了時實和壓力，這其實就很nice了。
最後我想說一句，其實不論是pull還是push，又或是輪詢和長輪詢，其實都是一種理論或者說是一種思想，不單單是MQ的東西，就比如在Nacos中，也使用了push和長輪詢機制。但是這些理論在不同產品的具體實現，實現方式可能不太一樣，但都是大同小異，所以當你懂了這些思想，再看其它框架的原始碼，其實就很容易了。
最後的最後，我再說一句，終於***發年終獎了。。
本文如果對你有點幫助，還請幫忙點贊、在看、轉發、非常感謝。
往期熱門文章推薦
掃碼或者搜尋關注公眾號 三友的java日記 ，及時乾貨不錯過，公眾號致力於通過畫圖加上通俗易懂的語言講解技術，讓技術更加容易學習，回覆 面試 即可獲得一套面試真題。
