【RocketMQ】順序訊息實現原理
2022-11-22 06:01:36
全域性有序
在RocketMQ中，如果使訊息全域性有序，可以為Topic設定一個訊息佇列，使用一個生產者單執行緒傳送資料，消費者端也使用單執行緒進行消費，從而保證訊息的全域性有序，但是這種方式效率低，一般不使用。
區域性有序
假設一個Topic分配了兩個訊息佇列，生產者在傳送訊息的時候，可以對訊息設定一個路由ID，比如想保證一個訂單的相關訊息有序，那麼就使用訂單ID當做路由ID，在傳送訊息的時候，通過訂單ID對訊息佇列的個數取餘，根據取餘結果選擇訊息佇列，這樣同一個訂單的資料就可以保證傳送到一個訊息佇列中，消費者端使用MessageListenerOrderly處理有序訊息，這就是RocketMQ的區域性有序，保證訊息在某個訊息佇列中有序。
接下來看RoceketMQ原始碼中提供的順序訊息例子（稍微做了一些修改）：
生產者
public class Producer {
public static void main(String[] args) throws UnsupportedEncodingException {
try {
// 建立生產者
DefaultMQProducer producer = new DefaultMQProducer("生產者組");
// 啟動
producer.start();
// 建立TAG
String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
for (int i = 0; i < 100; i++) {
// 生成訂單ID
int orderId = i % 10;
// 建立訊息
Message msg =
new Message("TopicTest", tags[i % tags.length], "KEY" + i,
("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
// 獲取訂單ID
Integer id = (Integer) arg;
// 對訊息佇列個數取餘
int index = id % mqs.size();
// 根據取餘結果選擇訊息要傳送給哪個訊息佇列
return mqs.get(index);
}
}, orderId); // 這裡傳入了訂單ID
System.out.printf("%s%n", sendResult);
}
producer.shutdown();
} catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
e.printStackTrace();
}
}
}
消費者
public class Consumer {
public static void main(String[] args) throws MQClientException {
// 建立消費者
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("消費者組");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
// 訂閱主題
consumer.subscribe("TopicTest", "TagA || TagC || TagD");
// 註冊訊息監聽器，使用的是MessageListenerOrderly
consumer.registerMessageListener(new MessageListenerOrderly() {
@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
context.setAutoCommit(true);
// 列印訊息
System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
return ConsumeOrderlyStatus.SUCCESS;
}
});
consumer.start();
System.out.printf("Consumer Started.%n");
}
}
從例子中可以看出生產者在傳送訊息的時候，通過訂單ID作為路由資訊，將同一個訂單ID的訊息傳送到了同一個訊息佇列中，保證同一個訂單ID的訊息有序，那麼消費者端是如何保證訊息的順序讀取呢？接下來就去看下原始碼。
順序訊息實現原理
在【RocketMQ】訊息的拉取一文中講到，消費者在啟動時會呼叫DefaultMQPushConsumerImpl的start方法：
public class DefaultMQPushConsumer extends ClientConfig implements MQPushConsumer {
/**
* 預設的訊息推播實現類
*/
protected final transient DefaultMQPushConsumerImpl defaultMQPushConsumerImpl;
/**
* 啟動
*/
@Override
public void start() throws MQClientException {
setConsumerGroup(NamespaceUtil.wrapNamespace(this.getNamespace(), this.consumerGroup));
// 啟動消費者
this.defaultMQPushConsumerImpl.start();
// ...
}
}
在DefaultMQPushConsumerImpl的start方法中，對訊息監聽器型別進行了判斷，如果型別是MessageListenerOrderly表示要進行順序消費，此時使用ConsumeMessageOrderlyService對ConsumeMessageService進行範例化，然後呼叫它的start方法進行啟動：
public class DefaultMQPushConsumerImpl implements MQConsumerInner {
// 訊息消費service
private ConsumeMessageService consumeMessageService;
public synchronized void start() throws MQClientException {
switch (this.serviceState) {
case CREATE_JUST:
// ...
// 如果是順序消費
if (this.getMessageListenerInner() instanceof MessageListenerOrderly) {
// 設定順序消費標記
this.consumeOrderly = true;
// 建立consumeMessageService，使用的是ConsumeMessageOrderlyService
this.consumeMessageService =
new ConsumeMessageOrderlyService(this, (MessageListenerOrderly) this.getMessageListenerInner());
} else if (this.getMessageListenerInner() instanceof MessageListenerConcurrently) {
this.consumeOrderly = false;
// 並行消費使用ConsumeMessageConcurrentlyService
this.consumeMessageService =
new ConsumeMessageConcurrentlyService(this, (MessageListenerConcurrently) this.getMessageListenerInner());
}
// 啟動ConsumeMessageService
this.consumeMessageService.start();
// ...
break;
// ...
}
// ...
}
}
加鎖定時任務
進入到ConsumeMessageOrderlyService的start方法中，可以看到，如果是叢集模式，會啟動一個定時加鎖的任務，週期性的對訂閱的訊息佇列進行加鎖，具體是通過呼叫RebalanceImpl的lockAll方法實現的：
public class ConsumeMessageOrderlyService implements ConsumeMessageService {
public void start() {
// 如果是叢集模式
if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())) {
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
// 週期性的執行加鎖方法
ConsumeMessageOrderlyService.this.lockMQPeriodically();
} catch (Throwable e) {
log.error("scheduleAtFixedRate lockMQPeriodically exception", e);
}
}
}, 1000 * 1, ProcessQueue.REBALANCE_LOCK_INTERVAL, TimeUnit.MILLISECONDS);
}
}
public synchronized void lockMQPeriodically() {
if (!this.stopped) {
// 進行加鎖
this.defaultMQPushConsumerImpl.getRebalanceImpl().lockAll();
}
}
}
為什麼叢集模式下需要加鎖？
因為廣播模式下，訊息佇列會分配給消費者下的每一個消費者，而在叢集模式下，一個訊息佇列同一時刻只能被同一個消費組下的某一個消費者進行，所以在廣播模式下不存在競爭關係，也就不需要對訊息佇列進行加鎖，而在叢集模式下，有可能因為負載均衡等原因將某一個訊息佇列分配到了另外一個消費者中，因此在叢集模式下就要加鎖，當某個訊息佇列被鎖定時，其他的消費者不能進行消費。
訊息佇列加鎖
在RebalanceImpl的lockAll方法中，首先從處理佇列表中獲取當前消費者訂閱的所有訊息佇列MessageQueue資訊，返回資料是一個MAP，key為broker名稱，value為broker下的訊息佇列，接著對MAP進行遍歷，處理每一個broker下的訊息佇列：
- 獲取broker名稱，根據broker名稱查詢broker的相關資訊；
- 構建加鎖請求，在請求中設定要加鎖的訊息佇列，然後將請求傳送給broker，表示要對這些訊息佇列進行加鎖；
- 加鎖請求返回的響應結果中包含了加鎖成功的訊息佇列，此時遍歷加鎖成功的訊息佇列，將訊息佇列對應的
ProcessQueue中的locked屬性置為true表示該訊息佇列已加鎖成功； - 處理加鎖失敗的訊息佇列，如果響應中未包含某個訊息佇列的資訊，表示此訊息佇列加鎖失敗，需要將其對應的
ProcessQueue物件中的locked屬性置為false表示加鎖失敗；
public abstract class RebalanceImpl {
public void lockAll() {
// 從處理佇列表中獲取broker對應的訊息佇列，key為broker名稱，value為broker下的訊息佇列
HashMap<String, Set<MessageQueue>> brokerMqs = this.buildProcessQueueTableByBrokerName();
// 遍歷訂閱的訊息佇列
Iterator<Entry<String, Set<MessageQueue>>> it = brokerMqs.entrySet().iterator();
while (it.hasNext()) {
Entry<String, Set<MessageQueue>> entry = it.next();
// broker名稱
final String brokerName = entry.getKey();
// 獲取訊息佇列
final Set<MessageQueue> mqs = entry.getValue();
if (mqs.isEmpty())
continue;
// 根據broker名稱獲取broker資訊
FindBrokerResult findBrokerResult = this.mQClientFactory.findBrokerAddressInSubscribe(brokerName, MixAll.MASTER_ID, true);
if (findBrokerResult != null) {
// 構建加鎖請求
LockBatchRequestBody requestBody = new LockBatchRequestBody();
// 設定消費者組
requestBody.setConsumerGroup(this.consumerGroup);
// 設定ID
requestBody.setClientId(this.mQClientFactory.getClientId());
// 設定要加鎖的訊息佇列
requestBody.setMqSet(mqs);
try {
// 批次進行加鎖，返回加鎖成功的訊息佇列
Set<MessageQueue> lockOKMQSet =
this.mQClientFactory.getMQClientAPIImpl().lockBatchMQ(findBrokerResult.getBrokerAddr(), requestBody, 1000);
// 遍歷加鎖成功的佇列
for (MessageQueue mq : lockOKMQSet) {
// 從處理佇列表中獲取對應的處理佇列物件
ProcessQueue processQueue = this.processQueueTable.get(mq);
// 如果不為空，設定locked為true表示加鎖成功
if (processQueue != null) {
if (!processQueue.isLocked()) {
log.info("the message queue locked OK, Group: {} {}", this.consumerGroup, mq);
}
// 設定加鎖成功標記
processQueue.setLocked(true);
processQueue.setLastLockTimestamp(System.currentTimeMillis());
}
}
// 處理加鎖失敗的訊息佇列
for (MessageQueue mq : mqs) {
if (!lockOKMQSet.contains(mq)) {
ProcessQueue processQueue = this.processQueueTable.get(mq);
if (processQueue != null) {
// 設定加鎖失敗標記
processQueue.setLocked(false);
log.warn("the message queue locked Failed, Group: {} {}", this.consumerGroup, mq);
}
}
}
} catch (Exception e) {
log.error("lockBatchMQ exception, " + mqs, e);
}
}
}
}
}
在【RocketMQ】訊息的拉取一文中講到，消費者需要先向Broker傳送拉取訊息請求，從Broker中拉取訊息，拉取訊息請求構建在RebalanceImpl的updateProcessQueueTableInRebalance方法中，拉取訊息的響應結果處理在PullCallback的onSuccess方法中，接下來看下順序消費時在這兩個過程中是如何處理的。
拉取訊息
上面已經知道，在使用順序訊息時，會週期性的對訂閱的訊息佇列進行加鎖，不過由於負載均衡等原因，有可能給當前消費者分配新的訊息佇列，此時可能還未來得及通過定時任務加鎖，所以消費者在構建訊息拉取請求前會再次進行判斷，如果processQueueTable中之前未包含某個訊息佇列，會先呼叫lock方法進行加鎖，lock方法的實現邏輯與lockAll基本一致，如果加鎖成功構建拉取請求進行訊息拉取，如果加鎖失敗，則跳過繼續處理下一個訊息佇列：
public abstract class RebalanceImpl {
private boolean updateProcessQueueTableInRebalance(final String topic, final Set<MessageQueue> mqSet,
final boolean isOrder) {
// ...
List<PullRequest> pullRequestList = new ArrayList<PullRequest>();
// 遍歷佇列集合
for (MessageQueue mq : mqSet) {
// 如果processQueueTable之前不包含當前的訊息佇列
if (!this.processQueueTable.containsKey(mq)) {
// 如果是順序消費，呼叫lock方法進行加鎖，如果加鎖失敗不往下執行，繼續處理下一個訊息佇列
if (isOrder && !this.lock(mq)) {
log.warn("doRebalance, {}, add a new mq failed, {}, because lock failed", consumerGroup, mq);
continue;
}
// ...
// 如果偏移量大於等於0
if (nextOffset >= 0) {
// 放入處理佇列表中
ProcessQueue pre = this.processQueueTable.putIfAbsent(mq, pq);
if (pre != null) {
log.info("doRebalance, {}, mq already exists, {}", consumerGroup, mq);
} else {
// 如果之前不存在，構建PullRequest，之後對請求進行處理，進行訊息拉取
log.info("doRebalance, {}, add a new mq, {}", consumerGroup, mq);
PullRequest pullRequest = new PullRequest();
pullRequest.setConsumerGroup(consumerGroup);
pullRequest.setNextOffset(nextOffset);
pullRequest.setMessageQueue(mq);
pullRequest.setProcessQueue(pq);
pullRequestList.add(pullRequest);
changed = true;
}
} else {
log.warn("doRebalance, {}, add new mq failed, {}", consumerGroup, mq);
}
}
}
// 新增訊息拉取請求
this.dispatchPullRequest(pullRequestList);
return changed;
}
public boolean lock(final MessageQueue mq) {
// 獲取broker資訊
FindBrokerResult findBrokerResult = this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(), MixAll.MASTER_ID, true);
if (findBrokerResult != null) {
// 構建加鎖請求
LockBatchRequestBody requestBody = new LockBatchRequestBody();
requestBody.setConsumerGroup(this.consumerGroup);
requestBody.setClientId(this.mQClientFactory.getClientId());
// 設定要加鎖的訊息佇列
requestBody.getMqSet().add(mq);
try {
// 傳送加鎖請求
Set<MessageQueue> lockedMq =
this.mQClientFactory.getMQClientAPIImpl().lockBatchMQ(findBrokerResult.getBrokerAddr(), requestBody, 1000);
for (MessageQueue mmqq : lockedMq) {
ProcessQueue processQueue = this.processQueueTable.get(mmqq);
// 如果加鎖成功設定成功標記
if (processQueue != null) {
processQueue.setLocked(true);
processQueue.setLastLockTimestamp(System.currentTimeMillis());
}
}
boolean lockOK = lockedMq.contains(mq);
log.info("the message queue lock {}, {} {}",
lockOK ? "OK" : "Failed",
this.consumerGroup,
mq);
return lockOK;
} catch (Exception e) {
log.error("lockBatchMQ exception, " + mq, e);
}
}
return false;
}
}
順序訊息消費
在PullCallback的onSuccess方法中可以看到，如果從Broker拉取到訊息，會呼叫ConsumeMessageService的submitConsumeRequest方法將訊息提交到ConsumeMessageService中進行消費：
public class DefaultMQPushConsumerImpl implements MQConsumerInner {
public void pullMessage(final PullRequest pullRequest) {
// ...
// 拉取訊息回撥函數
PullCallback pullCallback = new PullCallback() {
@Override
public void onSuccess(PullResult pullResult) {
if (pullResult != null) {
// 處理拉取結果
pullResult = DefaultMQPushConsumerImpl.this.pullAPIWrapper.processPullResult(pullRequest.getMessageQueue(), pullResult,
subscriptionData);
// 判斷拉取結果
switch (pullResult.getPullStatus()) {
case FOUND:
// ...
// 如果未拉取到訊息
if (pullResult.getMsgFoundList() == null || pullResult.getMsgFoundList().isEmpty()) {
// 將拉取請求放入到阻塞佇列中再進行一次拉取
DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
} else {
// ...
// 如果拉取到訊息，將訊息提交到ConsumeMessageService中進行消費
DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
pullResult.getMsgFoundList(),
processQueue,
pullRequest.getMessageQueue(),
dispatchToConsume);
// ...
}
// ...
}
}
}
};
}
}
前面知道順序消費時使用的是ConsumeMessageOrderlyService，首先在ConsumeMessageOrderlyService的建構函式中可以看到
初始化了一個訊息消費執行緒池，也就是說順序消費時也是開啟多執行緒進行消費的：
public class ConsumeMessageOrderlyService implements ConsumeMessageService {
public ConsumeMessageOrderlyService(DefaultMQPushConsumerImpl defaultMQPushConsumerImpl,
MessageListenerOrderly messageListener) {
// ...
// 設定訊息消費執行緒池
this.consumeExecutor = new ThreadPoolExecutor(
this.defaultMQPushConsumer.getConsumeThreadMin(),
this.defaultMQPushConsumer.getConsumeThreadMax(),
1000 * 60,
TimeUnit.MILLISECONDS,
this.consumeRequestQueue,
new ThreadFactoryImpl(consumeThreadPrefix));
this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("ConsumeMessageScheduledThread_"));
}
}
接下來看submitConsumeRequest方法，可以看到構建了ConsumeRequest物件，將拉取的訊息提交到了訊息消費執行緒池中進行消費：
public class ConsumeMessageOrderlyService implements ConsumeMessageService {
@Override
public void submitConsumeRequest(
final List<MessageExt> msgs,
final ProcessQueue processQueue,
final MessageQueue messageQueue,
final boolean dispathToConsume) {
if (dispathToConsume) {
// 構建ConsumeRequest
ConsumeRequest consumeRequest = new ConsumeRequest(processQueue, messageQueue);
this.consumeExecutor.submit(consumeRequest);
}
}
}
消費時的訊息佇列鎖
ConsumeRequest是ConsumeMessageOrderlyService的內部類，它有兩個成員變數，分別為MessageQueue訊息佇列和它對應的處理佇列ProcessQueue物件。
在run方法中，對訊息進行消費，處理邏輯如下：
- 判斷
ProcessQueue是否被刪除，如果被刪除終止處理； - 呼叫messageQueueLock的ftchLockObject方法獲取訊息佇列的物件鎖，然後使用synchronized進行加鎖，這裡加鎖的原因是因為順序消費使用的是執行緒池，可以設定多個執行緒同時進行消費，所以某個執行緒在進行訊息消費的時候要對訊息佇列加鎖，防止其他執行緒並行消費，破壞訊息的順序性；
- 如果是廣播模式、或者當前的訊息佇列已經加鎖成功（Locked置為true）並且加鎖時間未過期，開始對拉取的訊息進行遍歷：
- 如果是叢集模式並且訊息佇列加鎖失敗，呼叫tryLockLaterAndReconsume稍後重新進行加鎖；
- 如果是叢集模式並且訊息佇列加鎖時間已經過期，呼叫tryLockLaterAndReconsume稍後重新進行加鎖；
- 如果當前時間距離開始處理的時間超過了最大消費時間，呼叫submitConsumeRequestLater稍後重新進行處理；
- 獲取批次消費訊息個數，從ProcessQueue獲取訊息內容，如果訊息獲取不為空，新增訊息消費鎖，然後呼叫messageListener的consumeMessage方法進行訊息消費；
public class ConsumeMessageOrderlyService implements ConsumeMessageService {
class ConsumeRequest implements Runnable {
private final ProcessQueue processQueue; // 訊息佇列對應的處理佇列
private final MessageQueue messageQueue; // 訊息佇列
public ConsumeRequest(ProcessQueue processQueue, MessageQueue messageQueue) {
this.processQueue = processQueue;
this.messageQueue = messageQueue;
}
@Override
public void run() {
// 處理佇列如果已經被置為刪除狀態，跳過不進行處理
if (this.processQueue.isDropped()) {
log.warn("run, the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
return;
}
// 獲取訊息佇列的物件鎖
final Object objLock = messageQueueLock.fetchLockObject(this.messageQueue);
// 物件訊息佇列的物件鎖加鎖
synchronized (objLock) {
// 如果是廣播模式、或者當前的訊息佇列已經加鎖成功並且加鎖時間未過期
if (MessageModel.BROADCASTING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
|| (this.processQueue.isLocked() && !this.processQueue.isLockExpired())) {
final long beginTime = System.currentTimeMillis();
for (boolean continueConsume = true; continueConsume; ) {
// 判斷processQueue是否刪除
if (this.processQueue.isDropped()) {
log.warn("the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
break;
}
// 如果是叢集模式並且processQueue的加鎖失敗
if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
&& !this.processQueue.isLocked()) {
log.warn("the message queue not locked, so consume later, {}", this.messageQueue);
// 稍後進行加鎖
ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 10);
break;
}
// 如果是叢集模式並且訊息佇列加鎖時間已經過期
if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
&& this.processQueue.isLockExpired()) {
log.warn("the message queue lock expired, so consume later, {}", this.messageQueue);
// 稍後進行加鎖
ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 10);
break;
}
long interval = System.currentTimeMillis() - beginTime;
// 如果當前時間距離開始處理的時間超過了最大消費時間
if (interval > MAX_TIME_CONSUME_CONTINUOUSLY) {
// 稍後重新進行處理
ConsumeMessageOrderlyService.this.submitConsumeRequestLater(processQueue, messageQueue, 10);
break;
}
// 批次消費訊息個數
final int consumeBatchSize =
ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize();
// 獲取訊息內容
List<MessageExt> msgs = this.processQueue.takeMessages(consumeBatchSize);
defaultMQPushConsumerImpl.resetRetryAndNamespace(msgs, defaultMQPushConsumer.getConsumerGroup());
if (!msgs.isEmpty()) {
// ...
try {
// 加消費鎖
this.processQueue.getConsumeLock().lock();
if (this.processQueue.isDropped()) {
log.warn("consumeMessage, the message queue not be able to consume, because it's dropped. {}",
this.messageQueue);
break;
}
// 消費訊息
status = messageListener.consumeMessage(Collections.unmodifiableList(msgs), context);
} catch (Throwable e) {
log.warn(String.format("consumeMessage exception: %s Group: %s Msgs: %s MQ: %s",
RemotingHelper.exceptionSimpleDesc(e),
ConsumeMessageOrderlyService.this.consumerGroup,
msgs,
messageQueue), e);
hasException = true;
} finally {
// 釋放訊息消費鎖
this.processQueue.getConsumeLock().unlock();
}
// ...
ConsumeMessageOrderlyService.this.getConsumerStatsManager()
.incConsumeRT(ConsumeMessageOrderlyService.this.consumerGroup, messageQueue.getTopic(), consumeRT);
continueConsume = ConsumeMessageOrderlyService.this.processConsumeResult(msgs, status, context, this);
} else {
continueConsume = false;
}
}
} else {
if (this.processQueue.isDropped()) {
log.warn("the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
return;
}
ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 100);
}
}
}
}
}
MessageQueueLock中使用了ConcurrentHashMap儲存每個訊息佇列對應的物件鎖，物件鎖實際上是一個Object類的物件，從Map中獲取訊息佇列的物件鎖時，如果物件鎖不存在，則新建一個Object物件，並放入Map集合中：
public class MessageQueueLock {
private ConcurrentMap<MessageQueue, Object> mqLockTable =
new ConcurrentHashMap<MessageQueue, Object>();
public Object fetchLockObject(final MessageQueue mq) {
// 獲取訊息佇列對應的物件鎖，也就是一個Object型別的物件
Object objLock = this.mqLockTable.get(mq);
// 如果獲取尾款
if (null == objLock) {
// 建立物件
objLock = new Object();
// 加入到Map中
Object prevLock = this.mqLockTable.putIfAbsent(mq, objLock);
if (prevLock != null) {
objLock = prevLock;
}
}
return objLock;
}
}
訊息消費鎖
ProcessQueue中持有一個訊息消費鎖，消費者呼叫consumeMessage進行訊息前，會新增消費鎖，上面已經知道在處理拉取到的訊息時就已經呼叫messageQueueLock的fetchLockObject方法獲取訊息佇列的物件鎖然後使用syncronized對其加鎖，那麼為什麼在消費之前還要再加一個消費鎖呢？
public class ProcessQueue {
// 訊息消費鎖
private final Lock consumeLock = new ReentrantLock();
public Lock getConsumeLock() {
return consumeLock;
}
}
這裡講一個小技巧，如果在檢視原始碼的時候對某個方法有疑問，可以檢視一下這個方法在哪裡被呼叫了，結合呼叫處的程式碼處理邏輯進行猜測。
那麼就來看下getConsumeLock在哪裡被呼叫了，可以看到除了C的run方法中呼叫了之外，RebalancePushImpl中的removeUnnecessaryMessageQueue方法也呼叫了getConsumeLock方法：
removeUnnecessaryMessageQueue方法從名字上可以看出，是移除不需要的訊息佇列，RebalancePushImpl是與負載均衡相關的類，所以猜測有可能在負載均衡時，需要移除某個訊息佇列，那麼消費者在進行消費的時候就要獲取ProcessQueue的consumeLock進行加鎖，防止正在消費的過程中，消費佇列被移除：
public class RebalancePushImpl extends RebalanceImpl {
@Override
public boolean removeUnnecessaryMessageQueue(MessageQueue mq, ProcessQueue pq) {
this.defaultMQPushConsumerImpl.getOffsetStore().persist(mq);
this.defaultMQPushConsumerImpl.getOffsetStore().removeOffset(mq);
// 如果是順序消費並且是集模式
if (this.defaultMQPushConsumerImpl.isConsumeOrderly()
&& MessageModel.CLUSTERING.equals(this.defaultMQPushConsumerImpl.messageModel())) {
try {
// 進行加鎖
if (pq.getConsumeLock().tryLock(1000, TimeUnit.MILLISECONDS)) {
try {
return this.unlockDelay(mq, pq);
} finally {
pq.getConsumeLock().unlock();
}
} else {
log.warn("[WRONG]mq is consuming, so can not unlock it, {}. maybe hanged for a while, {}",
mq,
pq.getTryUnlockTimes());
pq.incTryUnlockTimes();
}
} catch (Exception e) {
log.error("removeUnnecessaryMessageQueue Exception", e);
}
return false;
}
return true;
}
}
不過在消費者在消費訊息前已經對佇列進行了加鎖，負載均衡的時候為什麼不使用佇列鎖而要使用消費鎖？
這裡應該是為了減小鎖的粒度，因為消費者在對訊息佇列加鎖後，還進行了一系列的判斷，校驗都成功之後從處理佇列中獲取訊息內容，之後才開始消費訊息，如果負載均衡使用訊息佇列鎖就要等待整個過程完成才有可能加鎖成功，這樣顯然會降低效能，而如果使用訊息消費鎖，就可以減少等待的時間，並且消費者在進行訊息消費前也會判斷ProcessQueue是否被移除，所以只要保證consumeMessage方法在執行的過程中，ProcessQueue不被移除即可。
總結
目前一共涉及了三把鎖，它們分別對應不同的情況：
向Broker申請的訊息佇列鎖
叢集模式下一個訊息佇列同一時刻只能被同一個消費組下的某一個消費者進行，為了避免負載均衡等原因引起的變動，消費者會向Broker傳送請求對訊息佇列進行加鎖，如果加鎖成功，記錄到訊息佇列對應的ProcessQueue中的locked變數中，它是boolean型別的：
public class ProcessQueue {
private volatile boolean locked = false;
}
消費者處理拉取訊息時的訊息佇列鎖
消費者在處理拉取到的訊息時，由於可以開啟多執行緒進行處理，所以處理訊息前通過MessageQueueLock中的mqLockTable獲取到了訊息佇列對應的鎖，鎖住要處理的訊息佇列，這裡加訊息佇列鎖主要是處理多執行緒之間的競爭：
public class MessageQueueLock {
private ConcurrentMap<MessageQueue, Object> mqLockTable =
new ConcurrentHashMap<MessageQueue, Object>();
訊息消費鎖
消費者在呼叫consumeMessage方法之前會加消費鎖，主要是為了避免在消費訊息時，由於負載均衡等原因，ProcessQueue被刪除：
public class ProcessQueue {
private final Lock consumeLock = new ReentrantLock();
}
參考
丁威、周繼鋒《RocketMQ技術內幕》
RocketMQ版本：4.9.3