前言

小夥伴們，國慶都過的開心嗎？國慶後的第一個工作日是不是很多小夥伴還沉浸在假期的心情中，沒有工作狀態呢？

那王子今天和大家聊一聊RocketMQ的消費者是如何獲取訊息的，通過學習知識來找回狀態吧。

廢話不多說，我們開始吧。

消費者組

首先我們瞭解一個概念，什麼是消費者組。

消費者組你就可以把它理解為，給一組消費者起一個名字。

假設我們有一個訂單Topic名字是OrderTopic，然後庫存系統和積分系統都要消費這個Topic中的資料，我們分別給庫存系統和積分系統起一個消費組名字：stock_consumer_group、score_consumer_group。

設定消費者組名字是在程式碼中實現的，如下：

 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("stock_consumer_group");

比如我們的庫存系統提供了2臺機器，每臺機器上的消費者組名字都是stock_consumer_group，那麼這2臺機器就是一個消費者組。

大體結構如上圖所示，那麼當訂單系統傳送訊息到OrderTopic中後，庫存系統和積分系統是如何進行消費的呢？

預設情況下，這條訊息傳送到Broker後，庫存系統和積分系統都會拉取這條訊息，而且庫存系統的兩臺機器中只有一臺會消費到這條訊息，積分系統也一樣。

這就是消費組的概念，不同的系統設定不同的消費組，如果不同的消費組訂閱了同一個Topic，那麼對於Topic中的一條訊息，每個消費組都會獲取到這條訊息。

叢集模式和廣播模式

接下來我們思考一個問題，對於消費者組而言，當它獲取到一條訊息後，假設消費者組內有多臺機器，那麼到底是隻有一臺機器獲取到訊息，還是所有機器都獲取到訊息呢？

這其實是消費的兩種模式，叢集模式和廣播模式。

預設情況下我們都是使用的叢集模式，也就是說消費者組收到訊息後，只有其中的一臺機器會接收到訊息。

我們可以手動指定為廣播模式。

consumer.setMessageModel(MessageModel.BROADCASTING)

指定為廣播模式後，消費者組內的每臺機器都會收到這條訊息。

具體要根據業務場景選擇消費模式。

MessageQueue與消費者的關係

接著我們想一下，對於一個Topic下的多個MessageQueue，消費者組中的多臺機器是如何消費的呢？

這部分內容底層實現是很複雜的，我們可以簡單的理解為它會均勻的將多個MessageQueue分配給消費者組中的多臺機器消費。

舉個例子，假如我們的OrderTopic有四個MessageQueue，這4個MessageQueue分佈在兩臺MasterBroker上，每個MasterBroker上有兩個MessageQueue。

然後庫存系統作為一個消費者組有兩臺機器，那麼最好的分配方式就是每臺消費者機器負責兩個MessageQueue，這樣就實現了機器的負載消費，示意圖如下：

所以我們可以大致的認為，一個Topic中的多個MessageQueue會被均勻的分佈給一個消費者組中的多臺機器進行消費，這裡要注意一點，一個MessageQueue只能被一臺消費者機器消費，但是一臺消費者機器可以同時負責處理多個MessageQueue。

那麼當消費者組中的機器數量發生變化時，是怎麼處理的。

機器數量發生變化一般就兩種情況，一種是有機器宕機了，另一種是增加機器進行叢集擴容了。

其實這種情況下是會進行rebalance環節的，也就是會重新分配每個消費者機器要處理的MessageQueue。

Push模式和Pull模式

不知道小夥伴們還記不記得，在之前的文章RocketMQ的傳送模式和消費模式中，我們已經用程式碼說明了消費者的兩種消費模式：Push和Pull，當時只提供了Push消費的程式碼，而沒有提供Pull消費的程式碼。

其實這兩種模式本質上是一樣的，都是消費者主動發出請求到Broker上拉取訊息。

Push模式的底層也是通過消費者主動拉取的方式來實現的，只不過它的名字叫Push而已，意思是Broker儘可能實時的推送訊息給消費者。

我們一般在使用RocketMQ的時候，消費模式基本都是使用的Push模式，因為Pull模式真的使用起來程式碼特別複雜，而且Push模式的底層還是Pull模式，只是對時效性有了更好的支援。

Push模式大體實現思路是這樣的：當消費者傳送請求到Broker拉取訊息的時候，如果有新的訊息可以消費，會立馬返回訊息到消費者進行消費，消費後會接著傳送請求到Broker拉取訊息。

也就說Push模式下，處理完一批訊息後會理解再發送請求給Broker拉取下一批訊息，所以時效性更好，看起來就像是Broker在實時推送訊息。

當請求傳送到Broker發現沒有需要消費的訊息時，就會讓請求執行緒掛起，預設掛起15秒，然後會有另一個後臺執行緒每隔一段時間判斷一下是否有新訊息需要消費，一旦發現了新的訊息，就會去喚醒掛起的執行緒，將訊息返回給消費者進行消費，然後消費完畢再次傳送請求拉取訊息。

這一部分的原始碼實現是很複雜的，我們只要瞭解它的核心思路就可以了。就算是Push模式，本質上也是對Pull模式的一種封裝。

Broker如何讀取訊息返回給消費者

接下來我們來聊聊Broker是如何讀取訊息返回給消費者的。之前的文章深入研究Broker是如何持久化的中我們已經知道了Broker是如何持久化訊息的，小夥伴們可以複習一下。

那麼當消費者傳送請求到Broker中拉取訊息時，假設是第一次拉取，就會從MessageQueue中的第一條訊息開始拉取。

如何定位到第一條訊息的位置呢，首先Broker會找到MessageQueue對應的ConsumerQueue，從裡面找到這條訊息的offset，然後通過offset去CommitLog中讀取訊息資料，把訊息返回給消費者。

當消費者消費完這條訊息後，會提交一個消費的進度給Broker，Broker會記錄下一個ConsumerOffset來標記我們的消費進度。

下次消費者再去這個MessageQueue中拉取訊息時，就會從記錄的消費位置繼續拉取訊息，而不用從頭獲取了。

總結

好了，到這裡本篇文章就結束了。

今天主要和大家一起討論了一下RocketMQ消費者的拉取和消費過程，也是國慶假期後的第一篇文章。

沒有從國慶中收回心的小夥伴們（ps：王子也一樣沒有進入狀態(｀・ω・´)）就與王子一起通過學習找回狀態吧。

往期文章推薦：

什麼是訊息中介軟體？主要作用是什麼？

常見的訊息中介軟體有哪些？你們是怎麼進行技術選型的？

你懂RocketMQ 的架構原理嗎？

聊一聊RocketMQ的註冊中心NameServer

Broker的主從架構是怎麼實現的？

RocketMQ生產部署架構如何設計

RabbitMQ和Kafka的高可用叢集原理

RocketMQ的傳送模式和消費模式

討論一下秒殺系統的技術難點與解決方案

秒殺系統中的扣減庫存和流量削峰

深入研究RocketMQ生產者傳送訊息的底層原理

深入研究Broker是如何持久化的

Dledger是如何實現主從自動切換的