前言

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

那王子今天和大家聊一聊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是如何實現主從自動切換的

長按識別二維碼，了解更多