RocketMQ延迟消息简明介绍

private static final long FIRST_DELAY_TIME = 1000L;

private static final long DELAY_FOR_A_WHILE = 100L;

private static final long DELAY_FOR_A_PERIOD = 10000L;

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

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

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

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

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);
    }
}

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

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