一次 RocketMQ 順序消費延遲的問題定位

一次 RocketMQ 順序消費延遲的問題定位

問題背景與現象

昨晚收到了應用報警，發現線上某個業務消費消息延遲了 54s 多（從消息發送到MQ 到被消費的間隔）：

2021-06-30T23:12:46.756 message processing is incredibly delayed! (Current delay time: 54725, incredible delay count in 10 seconds: 5677)

查看 RocketMQ 的監控，發現確實發生了比較多的消息積壓：
image

從 RocketMQ-Console 上面查看 Topic 的消費者：
image

這個 Topic，業務要求是需要有序的。所以在發送的時候，指定了業務 Key，並且消費的時候，使用的是順序消費模式。

我們使用了 RocketMQ 集群，有三個 Broker，對於這個 Topic，每個 Broker 上面都有 8 個 ReadQueue 和 WriteQueue。這裡簡單提一下 ReadQueue 和 WriteQueue 的意思：

在 RocketMQ 中，消息發送時使用 WriteQueue 個數返迴路由資訊，而消息消費時按照 ReadQueue 個數返迴路由資訊。在物理文件層面，只有 WriteQueue 才會創建文件。舉個例子：設置 WriteQueueNum = 8，ReadQueueNum = 4，會創建 8 個文件夾，代表 0 1 2 3 4 5 6 7 這 8 個隊列，但在消息消費時，路由資訊只返回 4，在具體拉取消息時，就只會消費0 1 2 3 這 4 個隊列中的消息，4 5 6 7 壓根就沒有被消費。反過來，如果設置 WriteQueueNum = 4，ReadQueueNum = 8，在生產消息時只會往0 1 2 3中生產消息，消費消息時則會從0 1 2 3 4 5 6 7 所有的隊列中消費，當然 4 5 6 7中壓根就沒有消息 ，假設消費是 Group 消費，Group 中有兩個消費者，事實上只有第一個消費者在真正的消費消息(0 1 2 3)，第二個消費者壓根就消費不到消息(4 5 6 7)。一般我們都會設置這兩個值相同，只有在需要縮容 topic 的隊列數量的時候，才會設置他們不同。

問題分析

首先聯想到的是，是否是消費執行緒卡住了呢？執行緒卡住一般因為：

  1. 發生了 Stop-the-wolrd：
  2. GC 導致
  3. 其他 safepoint 原因導致（例如 jstack，定時進入 safepoint 等等，參考我的這篇文章JVM相關 – SafePoint 與 Stop The World 全解）
  4. 執行緒處理消息時間過長，可能有鎖獲取不到，可能卡在某些 IO

採集當時的 JFR（關於 JFR，請參考我的另一系列JFR全解），發現：

  1. 在這個時間段並沒有發生停滯時間很長的 GC 以及其他 Stop-the-world 的 safepoint 事件：

image

image

  1. 在這段時間，執行緒是 park 的，並且堆棧顯示是消費執行緒並沒有消息可以消費：
    image

既然應用並沒有什麼問題，我們來看看 RocketMQ 是否有什麼問題。一般的 RocketMQ Broker 的日誌我們關心：

  1. 消息持久化的時間消耗統計，如果這裡發生異常，我們需要調優 Java MMAP 相關的參數，請參考：
  2. 消息持久化異常，查看 storeerr.log
  3. 鎖異常，查看 lock.log

那究竟應該去看哪一個 broker 呢？之前提到了，發送到這個 Topic 是指定了 hashKey 的，通過消息的 hashKey 我們可以定位到是哪個 broker：

int hashCode = "我們的hashKey".hashCode();
log.info("{}", Math.abs(hashCode % 24));

我們找到了消息的 hashKey，通過上面的程式碼，結果是 20，也就是隊列 20，通過前面的描述，我們知道每個 broker 是 8 個隊列，20 對應的就是 broker-2 上面的隊列，也就是 broker-2 queueId = 5 這個隊列。我們來查看 broker-2 上面的日誌定位問題。

我們發現 lock.log 裡面有異常，如下所示，類似的有很多條，並且持續了 54s 左右，和執行緒 park 時間比較吻合，也和消息延遲比較吻合：

2021-07-01 07:11:47 WARN AdminBrokerThread_10 - tryLockBatch, message queue locked by other client. Group: 消費group OtherClientId: 10.238.18.6@29 NewClientId: 10.238.18.122@29 MessageQueue [topic=消息topic, brokerName=broker-2, queueId=5]

這個日誌的意思是，10.238.18.122@29 這個實例嘗試鎖住 queueId = 5 失敗，因為 10.238.18.6@29 正在持有這個鎖。那麼為什麼會發生這種情況呢？

RocketMQ 多隊列順序消費的原理

RocketMQ 想要實現多隊列順序消費，首先需要指定 hashKey，通過 hashKey 消息會被放入特定的隊列，消費者消費這個隊列的時候，如果指定了順序消費，是單執行緒消費的，這樣就保證了同一隊列內有序。

那麼是如何保證每個隊列是單執行緒消費的呢？每個 Broker 維護一個：

private final ConcurrentMap<String/* group */, ConcurrentHashMap<MessageQueue, LockEntry>> mqLockTable =
        new ConcurrentHashMap<String, ConcurrentHashMap<MessageQueue, LockEntry>>(1024);

他是一個 ConcurrentMap<消費組名稱, ConcurrentHashMap<消息隊列, 鎖對象>>。鎖對象 LockEntry 包括：

RebalanceLockManager.java:

//讀取 rocketmq.broker.rebalance.lockMaxLiveTime 這個環境變數，默認 60s
private final static long REBALANCE_LOCK_MAX_LIVE_TIME = Long.parseLong(System.getProperty(
        "rocketmq.broker.rebalance.lockMaxLiveTime", "60000"));
static class LockEntry {
    //RocketMQ 客戶端唯一 id
    private String clientId;
    private volatile long lastUpdateTimestamp = System.currentTimeMillis();
    
    //省略getter setter
    
    public boolean isLocked(final String clientId) {
        boolean eq = this.clientId.equals(clientId);
        return eq && !this.isExpired();
    }

    public boolean isExpired() {
        // 在 REBALANCE_LOCK_MAX_LIVE_TIME 這麼長時間後過期
        boolean expired =
            (System.currentTimeMillis() - this.lastUpdateTimestamp) > REBALANCE_LOCK_MAX_LIVE_TIME;

        return expired;
    }
}

RocketMQ 客戶端發送 LOCK_BATCH_MQ 請求到 Broker 上面，Broker 會將客戶端請求封裝成為 LockEntry 並嘗試更新這個 Map，如果更新成功就是獲取到了鎖，如果失敗則沒有獲取這個鎖。Broker 的詳細更新邏輯是（感興趣可以查看，也可以直接跳過，不影響理解，後面有便於理解的圖片）：

public boolean tryLock(final String group, final MessageQueue mq, final String clientId) {
    //判斷沒有已經鎖住
    if (!this.isLocked(group, mq, clientId)) {
        try {
            //獲取鎖，這個鎖是實例內的，因為每個 broker 維護自己的隊列鎖表，並不共享
            this.lock.lockInterruptibly();
            try {
                //嘗試獲取，判斷是否存在，存在就判斷是否過期
                ConcurrentHashMap<MessageQueue, LockEntry> groupValue = this.mqLockTable.get(group);
                if (null == groupValue) {
                    groupValue = new ConcurrentHashMap<>(32);
                    this.mqLockTable.put(group, groupValue);
                }
                
                LockEntry lockEntry = groupValue.get(mq);
                if (null == lockEntry) {
                    lockEntry = new LockEntry();
                    lockEntry.setClientId(clientId);
                    groupValue.put(mq, lockEntry);
                    log.info("tryLock, message queue not locked, I got it. Group: {} NewClientId: {} {}",
                        group,
                        clientId,
                        mq);
                }

                if (lockEntry.isLocked(clientId)) {
                    lockEntry.setLastUpdateTimestamp(System.currentTimeMillis());
                    return true;
                }

                String oldClientId = lockEntry.getClientId();

                if (lockEntry.isExpired()) {
                    lockEntry.setClientId(clientId);
                    lockEntry.setLastUpdateTimestamp(System.currentTimeMillis());
                    log.warn(
                        "tryLock, message queue lock expired, I got it. Group: {} OldClientId: {} NewClientId: {} {}",
                        group,
                        oldClientId,
                        clientId,
                        mq);
                    return true;
                }
                //這裡就是我們剛剛看到的日誌
                log.warn(
                    "tryLock, message queue locked by other client. Group: {} OtherClientId: {} NewClientId: {} {}",
                    group,
                    oldClientId,
                    clientId,
                    mq);
                return false;
            } finally {
                this.lock.unlock();
            }
        } catch (InterruptedException e) {
            log.error("putMessage exception", e);
        }
    } else {

    }

    return true;
}
//判斷是否是已經鎖住了
private boolean isLocked(final String group, final MessageQueue mq, final String clientId) {
    //通過消費組名稱獲取
    ConcurrentHashMap<MessageQueue, LockEntry> groupValue = this.mqLockTable.get(group);
    //如果不為 null
    if (groupValue != null) {
        //嘗試獲取 lockEntry，看是否存在
        LockEntry lockEntry = groupValue.get(mq);
        if (lockEntry != null) {
            //如果存在，判斷是否過期
            boolean locked = lockEntry.isLocked(clientId);
            if (locked) {
                lockEntry.setLastUpdateTimestamp(System.currentTimeMillis());
            }

            return locked;
        }
    }

    return false;
}

每個 MQ 客戶端，會定時發送 LOCK_BATCH_MQ 請求，並且在本地維護獲取到鎖的所有隊列：

ProcessQueue.java:

//定時發送 **LOCK_BATCH_MQ** 間隔
public final static long REBALANCE_LOCK_INTERVAL = Long.parseLong(System.getProperty("rocketmq.client.rebalance.lockInterval", "20000"));

ConsumeMessageOrderlyService.java:

if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())) {
    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
            ConsumeMessageOrderlyService.this.lockMQPeriodically();
        }
    }, 1000 * 1, ProcessQueue.REBALANCE_LOCK_INTERVAL, TimeUnit.MILLISECONDS);
}

流程圖如下所示：

image

ConsumeMessageOrderlyService 在關閉的時候，會 unlock 所有的隊列：

public void shutdown() {
    this.stopped = true;
    this.scheduledExecutorService.shutdown();
    this.consumeExecutor.shutdown();
    if (MessageModel.CLUSTERING.equals(this.defaultMQPushConsumerImpl.messageModel())) {
        this.unlockAllMQ();
    }
}

問題出現原因

我們這裡客戶端定時發送 LOCK_BATCH_MQ 間隔是默認的 20s， Broker 端鎖過期的時間也是默認的 60s。

我們的集群容器編排使用了 k8s，並且有實例遷移的功能。在集群壓力大的時候，自動擴容新的 Node （可以理解為虛擬機）並將創建新的服務實例部署上去。集群某些服務壓力小的時候，某些服務實例會縮容下去，這時候就不需要那麼多 Node 了，就會回收一部分 Node，但是被回收的 Node 上面還有不能縮容的服務實例，這時候就需要將這些服務實例遷移到其他 Node 上面。這裡我們的業務實例就是發生了這個情況。

在問題出現的時候，發生了遷移，老的實例被關閉，但是沒有等待 ConsumeMessageOrderlyService#shutdown 的執行，導致鎖沒有被主動釋放，而是等待 60s 的鎖過期時間後，新的實例才拿到隊列鎖開始消費。

問題解決

  1. 在下個版本，加入針對 RocketMQ 客戶端的優雅關閉邏輯
  2. 所有服務實例（RocketMQ 客戶端）配置 rocketmq.client.rebalance.lockInterval 縮短心跳時間（5s），RocketMQ Broker 配置 rocketmq.broker.rebalance.lockMaxLiveTime 縮短過期時間（例如 15s），但是保持過期時間是心跳時間的 3 倍（集群中的 3 倍設計公理）

微信搜索「我的編程喵」關注公眾號，每日一刷，輕鬆提升技術，斬獲各種offer：

image

Tags:

VirMach 便宜 VPS

QNews

QNews