基於Redis實現延時佇列的優化方案小結

zadd delayqueue 1614608094000 taskinfo

zrangebyscore delayqueue 0 timestamp limit 0 1000

@XxlScheduled(cron = "0/5 * * * * ?", name = "scan business1 delayqueue")
public void scanBusiness1() {
    // 某業務邏輯的zset延遲佇列對應的key
    String zsetKey = "delayqueue:business1";
    while (true) {
        // 篩選score值小於當前時間戳的元素，一次最多拉取1000條
        Set<String> tasks = stringRedisTemplate.opsForZSet().rangeByScore(zsetKey, 0, System.currentTimeMillis(), 0, 1000);
        if (CollectionUtils.isEmpty(tasks)) {
            // 當前時間下已沒有需要執行的任務，結束本次掃描
            return;
        }
        for (String task : tasks) {
            // 先刪除，再執行，確保多執行緒環境下執行的唯一性
            Boolean delete = stringRedisTemplate.delete(task);
            if (delete) {
                // 刪除成功後，將其再傳送到指定MQ非同步處理，將「獲取任務」與「執行任務」分離解耦
                rabbitTemplate.convertAndSend("exchange_business1", "routekey_business1", task);
            }
        }
    }
}

@Scheduled(cron = "0/5 * * * * ?")
public void scanBusiness1() {
    // 佇列數量M，考慮動態設定，保持和機器數量基本一致
    int M = 10;
    // redis自增key，用於負載均衡
    String incrKey = "incrkey:delayqueue:business1";
    // 每臺機器執行時，從不同的zset中拉取訊息，儘量確保不同機器存取不同zset
    String zsetKey = "delayqueue:business1:" + (stringRedisTemplate.opsForValue().increment(incrKey) % M);
    while (true) {
        // 此處邏輯和前面程式碼一致，省略。。。
    }
}

local zsetKey = 'delayqueue'
local timestamp = 1614608094000
local items = redis.call('zrangebyscore',zsetKey,0,timestamp,'limit',0,1)
if #items == 0 then
    return ''
else
    redis.call('zremrangebyrank',zsetKey,0,0)
    return items[1]
end

/**
 * 基於ZSET實現訊息延遲處理，score儲存執行時間點，到達時間點即會向指定佇列傳送該訊息；
 * 定義一個繼承本類的bean即可；
 */
public abstract class AbstractDelayedMsgScanTrigger implements Runnable, DisposableBean {

    private static final RedisScript<String> TRY_GET_AND_DEL_SCRIPT;
    static {
        // 獲取並刪除的lua指令碼，使用spring提供的api
        String sb = "local items = redis.call('zrangebyscore',KEYS[1],0,ARGV[1],'limit',0,1)n" +
                "if #items == 0 thenn" +
                "treturn ''n" +
                "elsen" +
                "tredis.call('zremrangebyrank',KEYS[1],0,0)n" +
                "treturn items[1]n" +
                "end";
        // 自有工具類，只要能建立出spring包下的 RedisScript 的實現類物件均可
        TRY_GET_AND_DEL_SCRIPT = RedisScriptHelper.createScript(sb, String.class);
    }

    private final ThreadPoolExecutor EXECUTOR = new ThreadPoolExecutor(getThreadNum(), getThreadNum(),
            0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), new NamedThreadFactory(getThreadNamePrefix()));
    private volatile boolean quit = false;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void startScan() {
        // bean構建完成後，啟動若干執行執行緒
        int threadNum = getThreadNum();
        for (int i = 0; i < threadNum; i++) {
            EXECUTOR.execute(this);
        }
    }

    @Override
    public void run() {
        while (!quit) {
            try {
                // 迴圈，採用lua獲取當前需要執行的任務並將其從redis中刪除
                String msg = stringRedisTemplate.execute(TRY_GET_AND_DEL_SCRIPT,
                        Lists.newArrayList(getDelayedMsgSourceKey()), String.valueOf(System.currentTimeMillis()));
                if (StringUtils.isNotBlank(msg)) {
                    // 訊息不為空，表示獲取任務成功，將其再傳送到指定MQ非同步處理，將「獲取任務」與「執行任務」分離解耦
                    rabbitTemplate.convertAndSend(getSendExchange(), getSendRoutingKey(), msg);
                } else {
                    // 獲取不到任務，表示當前時間點下暫無需要執行的任務，則執行緒休眠1S（可視情況調整）
                    SleepUtils.sleepSeconds(1);
                }
            } catch (Exception e) {
                Logs.MSG.error("delayed msg scan error, sourceKey:{}", getDelayedMsgSourceKey(), e);
            }
        }
    }

    @Override
    public void destroy() throws Exception {
        quit = true;
    }

    public void setQuit(boolean quit) {
        this.quit = quit;
    }

    /**
     * 獲取訊息的工作執行緒數量
     */
    protected abstract int getThreadNum();

    /**
     * 執行緒名稱字首，方便問題定位
     */
    protected abstract String getThreadNamePrefix();

    /**
     * 存放延遲訊息的ZSET佇列名
     */
    protected abstract String getDelayedMsgSourceKey();

    /**
     * 訊息到達執行時間點時將其通過指定 exchange 傳送到實時消費佇列中
     */
    protected abstract String getSendExchange();

    /**
     * 訊息到達執行時間點時將其通過指定 routingKey 傳送到實時消費佇列中
     */
    protected abstract String getSendRoutingKey();

}