RocketMQ的push消費方式實現的太聰明瞭

来源:https://www.cnblogs.com/zzyang/archive/2022/08/09/16565620.html
-Advertisement-
Play Games

大家好,我是三友,我又來了~~ 最近仍然暢游在RocketMQ的源碼中,這幾天剛好翻到了消費者的源碼,發現RocketMQ的對於push消費方式的實現簡直太聰明瞭,所以趁著我腦子裡還有點印象的時候,趕緊來寫一篇文章,來掰扯一下,防止過兩天就忘得一干二凈了。 MQ消費方式 消費方式就是指消費者如何從M ...


大家好,我是三友,我又來了~~

最近仍然暢游在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和長輪詢機制。但是這些理論在不同產品的具體實現,實現方式可能不太一樣,但都是大同小異,所以當你懂了這些思想,再看其它框架的源碼,其實就很容易了。

最後的最後,我再說一句,終於***發年終獎了。。

本文如果對你有點幫助,還請幫忙點贊、在看、轉發、非常感謝。

往期熱門文章推薦

三萬字盤點Spring/Boot的那些常用擴展點

一網打盡非同步神器CompletableFuture

@Async註解的坑,小心

萬字+28張圖帶你探秘小而美的規則引擎框

7000字+24張圖帶你徹底弄懂線程池

掃碼或者搜索關註公眾號 三友的java日記 ,及時乾貨不錯過,公眾號致力於通過畫圖加上通俗易懂的語言講解技術,讓技術更加容易學習,回覆 面試 即可獲得一套面試真題。


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • 1.ObjectPostProcessor 使用 前面介紹了 ObjectPostProcessor的基本概念。相信讀者已經明白,所有的過濾器都由對應的配置類來負責創建,配置類在將過濾器創建成功之後,會調用父類的postProcess方法,該 方法最終會調用到CompositeObjectPostP ...
  • 0. 前言 寫完這篇文章後發現自己對於 DP 的優化一竅不通,所以補了補 DP 的一些優化,寫篇 blog 總結一下。 1. 單調隊列/單調棧優化 1.2 演算法介紹 這應該算是最基礎的 DP 優化方法了。 顧名思義,單調隊列/單調棧優化 DP 就是保持容器內元素的單調性,以達成減少冗餘狀態的目的。 ...
  • 目錄 一.簡介 二.效果演示 三.源碼下載 四.猜你喜歡 零基礎 OpenGL (ES) 學習路線推薦 : OpenGL (ES) 學習目錄 >> OpenGL ES 基礎 零基礎 OpenGL (ES) 學習路線推薦 : OpenGL (ES) 學習目錄 >> OpenGL ES 轉場 零基礎 O ...
  • 前言 在Python中 in 操作符可以用於判斷某個元素是否存在於當前對象中,而對於不同的Python對象,使用 in 操作符的處理效率是不一樣的。 今天我們主要針對 4 種不同的Python數據類型進行學習:list列表、tuple元組、set集合、dict字典。 測試過程 我們用於測試的 4 種 ...
  • 多商戶商城系統,也稱為B2B2C(BBC)平臺電商模式多商家商城系統。可以快速幫助企業搭建類似拼多多/京東/天貓/淘寶的綜合商城。 多商戶商城系統支持商家入駐加盟,同時滿足平臺自營、旗艦店等多種經營方式。平臺可以通過收取商家入駐費,訂單交易服務費,提現手續費,簡訊通道費等多手段方式,實現整體盈利。 ...
  • 作者:清茶淡粥醬 鏈接:https://juejin.cn/post/7026734817853210661 Spring Security簡介 Spring Security 是一種高度自定義的安全框架,利用(基於)SpringIOC/DI和AOP功能,為系統提供了聲明式安全訪問控制功能,減少了為 ...
  • 精華筆記: 介面: 是一種數據類型(引用類型) 由interface定義 只能包含常量和抽象方法(所有數據預設都是常量,所有方法預設都是抽象的) 介面不能被實例化 介面是需要被實現/繼承的,實現/派生類:必須重寫所有抽象方法 一個類可以實現多個介面,用逗號分隔。若又繼承又實現時,應先繼承後實現。 接 ...
  • 1 同步鎖synchronized追本溯源 引言 提到synchronized,無論是在開發過程中和麵試過程中常常遇到的問題 synchronized;也算是重災區了 為什麼說是重災區? 因為他不像其他的代碼,是有源碼,可以查看的 synchronized是一個關鍵字。直接是找不到源代碼的 接下來 ...
一周排行
    -Advertisement-
    Play Games
  • Timer是什麼 Timer 是一種用於創建定期粒度行為的機制。 與標準的 .NET System.Threading.Timer 類相似,Orleans 的 Timer 允許在一段時間後執行特定的操作,或者在特定的時間間隔內重覆執行操作。 它在分散式系統中具有重要作用,特別是在處理需要周期性執行的 ...
  • 前言 相信很多做WPF開發的小伙伴都遇到過表格類的需求,雖然現有的Grid控制項也能實現,但是使用起來的體驗感並不好,比如要實現一個Excel中的表格效果,估計你能想到的第一個方法就是套Border控制項,用這種方法你需要控制每個Border的邊框,並且在一堆Bordr中找到Grid.Row,Grid. ...
  • .NET C#程式啟動閃退,目錄導致的問題 這是第2次踩這個坑了,很小的編程細節,容易忽略,所以寫個博客,分享給大家。 1.第一次坑:是windows 系統把程式運行成服務,找不到配置文件,原因是以服務運行它的工作目錄是在C:\Windows\System32 2.本次坑:WPF桌面程式通過註冊表設 ...
  • 在分散式系統中,數據的持久化是至關重要的一環。 Orleans 7 引入了強大的持久化功能,使得在分散式環境下管理數據變得更加輕鬆和可靠。 本文將介紹什麼是 Orleans 7 的持久化,如何設置它以及相應的代碼示例。 什麼是 Orleans 7 的持久化? Orleans 7 的持久化是指將 Or ...
  • 前言 .NET Feature Management 是一個用於管理應用程式功能的庫,它可以幫助開發人員在應用程式中輕鬆地添加、移除和管理功能。使用 Feature Management,開發人員可以根據不同用戶、環境或其他條件來動態地控制應用程式中的功能。這使得開發人員可以更靈活地管理應用程式的功 ...
  • 在 WPF 應用程式中,拖放操作是實現用戶交互的重要組成部分。通過拖放操作,用戶可以輕鬆地將數據從一個位置移動到另一個位置,或者將控制項從一個容器移動到另一個容器。然而,WPF 中預設的拖放操作可能並不是那麼好用。為瞭解決這個問題,我們可以自定義一個 Panel 來實現更簡單的拖拽操作。 自定義 Pa ...
  • 在實際使用中,由於涉及到不同編程語言之間互相調用,導致C++ 中的OpenCV與C#中的OpenCvSharp 圖像數據在不同編程語言之間難以有效傳遞。在本文中我們將結合OpenCvSharp源碼實現原理,探究兩種數據之間的通信方式。 ...
  • 一、前言 這是一篇搭建許可權管理系統的系列文章。 隨著網路的發展,信息安全對應任何企業來說都越發的重要,而本系列文章將和大家一起一步一步搭建一個全新的許可權管理系統。 說明:由於搭建一個全新的項目過於繁瑣,所有作者將挑選核心代碼和核心思路進行分享。 二、技術選擇 三、開始設計 1、自主搭建vue前端和. ...
  • Csharper中的表達式樹 這節課來瞭解一下表示式樹是什麼? 在C#中,表達式樹是一種數據結構,它可以表示一些代碼塊,如Lambda表達式或查詢表達式。表達式樹使你能夠查看和操作數據,就像你可以查看和操作代碼一樣。它們通常用於創建動態查詢和解析表達式。 一、認識表達式樹 為什麼要這樣說?它和委托有 ...
  • 在使用Django等框架來操作MySQL時,實際上底層還是通過Python來操作的,首先需要安裝一個驅動程式,在Python3中,驅動程式有多種選擇,比如有pymysql以及mysqlclient等。使用pip命令安裝mysqlclient失敗應如何解決? 安裝的python版本說明 機器同時安裝了 ...