RocketMQ延遲消息簡明介紹

Posted on by

前言

場景可以是這樣的，雙11搶手機，一個新手機4000-5000，到0點的時候，沖著興奮勁，搶到瞭。但是摸瞭摸錢包，又冷靜下來瞭，好像不是很必要換手機。就放在那裡沒有支付，過瞭30分鐘，自動取消瞭。這裡就是使用延遲消息的場景，當下單之後，向消息隊列發送一條延遲30分鐘消費的消息。等到30分鐘過瞭，然後消費消息，執行檢查任務，要是對應的訂單支付瞭，就什麼都不做，要是沒支付，就取消訂單。

RocketMQ的延遲消息是org.apache.rocketmq.broker.schedule.ScheduleMessageService類實現的

核心屬性

RMQ_SYS_SCHEDULE_TOPIC

在之前的版本中叫SCHEDULE_TOPIC，是系統內置的Topic,用來保存所有的定時消息。沒有執行的定時消息都會被保存在這個topic中。

FIRST_DELAY_TIME

第一次執行定時任務的延遲時間，默認是1秒。

private static final long FIRST_DELAY_TIME = 1000L;

DELAY_FOR_A_WHILE

第二次以及之後每次定時任務執行的間隔時間，默認100ms。

private static final long DELAY_FOR_A_WHILE = 100L;

DELAY_FOR_A_PERIOD

若是延遲消息投遞失敗，則在這個時間過後繼續投遞，默認10秒。

private static final long DELAY_FOR_A_PERIOD = 10000L;

delayLevelTable

這是保存延遲級別和延遲時間映射關系的地方

private final ConcurrentMap<Integer /* level */, Long/* delay timeMillis */> delayLevelTable =
    new ConcurrentHashMap<Integer, Long>(32);

offsetTable

保存延遲級別和對應的消費位點

private final ConcurrentMap<Integer /* level */, Long/* offset */> offsetTable =
    new ConcurrentHashMap<Integer, Long>(32);

核心方法

queueId2DelayLevel

將queueId轉換為延遲級別

public static int queueId2DelayLevel(final int queueId) {
    return queueId + 1;
}

delayLevel2QueueId

將延遲級別轉換為queueId

public static int delayLevel2QueueId(final int delayLevel) {
    return delayLevel - 1;
}

updateOffset

更新延遲消息topic的消費位點

private void updateOffset(int delayLevel, long offset) {
    this.offsetTable.put(delayLevel, offset);
    if (versionChangeCounter.incrementAndGet() % brokerController.getBrokerConfig().getDelayOffsetUpdateVersionStep() == 0) {
        long stateMachineVersion = brokerController.getMessageStore() != null ? brokerController.getMessageStore().getStateMachineVersion() : 0;
        dataVersion.nextVersion(stateMachineVersion);
    }
}

computeDeliverTimestamp

根據延遲消息級別和消息的存儲時間計算該延遲消息的投遞時間

public long computeDeliverTimestamp(final int delayLevel, final long storeTimestamp) {
    Long time = this.delayLevelTable.get(delayLevel);
    if (time != null) {
        return time + storeTimestamp;
    }
    return storeTimestamp + 1000;
}

start()

啟動延遲消息服務

shutdown()

關閉start方法中啟動的額timer任務

load()

加載消息的消費位點信息和全部的延遲級別信息。延遲級別信息默認如下。

private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

parseDelayLevel

格式化所有的延遲級別信息，保存到內存中。

到此這篇關於RocketMQ延遲消息簡明介紹的文章就介紹到這瞭,更多相關RocketMQ延遲消息內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

推薦閱讀：