基于Redis延遲隊列的實現(xiàn)代碼

更新時間：2021年05月13日 11:26:38 作者：程序員老鄭

在生活中很多時候都會用到延遲隊列，本文基于Redis延遲隊列的實現(xiàn)代碼，具有一定的參考價值，感興趣的小伙伴們可以參考一下

使用場景

工作中大家往往會遇到類似的場景：

1.對于紅包場景，賬戶 A 對賬戶 B 發(fā)出紅包通常在 1 天后會自動歸還到原賬戶。

2.對于實時支付場景，如果賬戶 A 對商戶 S 付款 100 元，5秒后沒有收到支付方回調(diào)將自動取消訂單。

解決方案分析

方案一：

采用通過定時任務采用數(shù)據(jù)庫/非關(guān)系型數(shù)據(jù)庫輪詢方案。

優(yōu)點：

1. 實現(xiàn)簡單，對于項目前期這樣是最容易的解決方案。

缺點：

1. DB 有效使用率低，需要將一部分的數(shù)據(jù)庫的QPS分配給 JOB 的無效輪詢。

2. 服務資源浪費，因為輪詢需要對所有的數(shù)據(jù)做一次 SCAN 掃描 JOB 服務的資源開銷很大。

方案二：

采用延遲隊列：

優(yōu)點：

1. 服務的資源使用率較高，能夠精確的實現(xiàn)超時任務的執(zhí)行。

2. 減少 DB 的查詢次數(shù)，能夠降低數(shù)據(jù)庫的壓力

缺點：

1. 對于延遲隊列來說本身設計比較復雜，目前沒有通用的比較好過的方案。

基于 Redis 的延遲隊列實現(xiàn)

基于以上的分析，我決定通過 Redis 來實現(xiàn)分布式隊列。

設計思路：

1. 第一步將需要發(fā)放的消息發(fā)送到延遲隊列中。

2. 延遲隊列將數(shù)據(jù)存入 Redis 的 ZSet 有序集合中score 為當前時間戳，member 存入需要發(fā)送的數(shù)據(jù)。

3. 添加一個 schedule 來進行對 Redis 有序隊列的輪詢。

4. 如果到達達到消息的執(zhí)行時間，那么就進行業(yè)務的執(zhí)行。

5. 如果沒有達到消息的執(zhí)行是將，那么消息等待下輪執(zhí)行。

實現(xiàn)步驟：

由于本處篇幅有限，所以只列舉部分代碼，完整的代碼可以在本文最后訪問 GitHub 獲取。由于本人閱歷/水平有限，如有建議/或更正歡迎留言或提問。先在此謝謝大家駐足閱讀 👏 👏 👏。

需要注意的問題：

單個 Redis 命令的執(zhí)行是原子性的，但 Redis 沒有在事務上增加任何維持原子性的機制，所以 Redis 事務的執(zhí)行并不是原子性的。

事務可以理解為一個打包的批量執(zhí)行腳本，但批量指令并非原子化的操作，中間某條指令的失敗不會導致前面已做指令的回滾，也不會造成后續(xù)的指令不做。

我們可以通過 Redis 的 eval 命令來執(zhí)行 lua 腳本來保證原子性實現(xiàn)Redis的事務。

實現(xiàn)步驟如下：

1. 延遲隊列接口

/**
 * 延遲隊列
 *
 * @author zhengsh
 * @date 2020-03-27
 */
public interface RedisDelayQueue<E extends DelayMessage> {

    String META_TOPIC_WAIT = "delay:meta:topic:wait";
    String META_TOPIC_ACTIVE = "delay:meta:topic:active";
    String TOPIC_ACTIVE = "delay:active:9999";
    /**
     * 拉取消息
     */
    void poll();

    /**
     * 推送延遲消息
     *
     * @param e
     */
    void push(E e);
}

2. 延遲隊列消息

/**
 * 消息體
 *
 * @author zhengsh
 * @date 2020-03-27
 */
@Setter
@Getter
public class DelayMessage {
    /**
     * 消息唯一標識
     */
    private String id;
    /**
     * 消息主題
     */
    private String topic = "default";
    /**
     * 具體消息 json
     */
    private String body;
    /**
     * 延時時間, 格式為時間戳: 當前時間戳 + 實際延遲毫秒數(shù)
     */
    private Long delayTime = System.currentTimeMillis() + 30000L;
    /**
     * 消息發(fā)送時間
     */
    private LocalDateTime createTime;
}

3. 延遲隊列實現(xiàn)

/**
 * 延遲隊列實現(xiàn)
 *
 * @author zhengsh
 * @date 2020-03-27
 */
@Component
public class RedisDelayQueueImpl<E extends DelayMessage> implements RedisDelayQueue<E> {
    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Override
    public void poll() {
        // todo
    }

    /**
     * 發(fā)送消息
     *
     * @param e
     */
    @SneakyThrows
    @Override
    public void push(E e) {
        try {
            String jsonStr = JSON.toJSONString(e);
            String topic = e.getTopic();
            String zkey = String.format("delay:wait:%s", topic);
            String u =
                    "redis.call('sadd', KEYS[1], ARGV[1])\n" +
                            "redis.call('zadd', KEYS[2], ARGV[2], ARGV[3])\n" +
                            "return 1";

            Object[] keys = new Object[]{serialize(META_TOPIC_WAIT), serialize(zkey)};
            Object[] values = new Object[]{ serialize(zkey), serialize(String.valueOf(e.getDelayTime())),serialize(jsonStr)};

            Long result = redisTemplate.execute((RedisCallback<Long>) connection -> {
                Object nativeConnection = connection.getNativeConnection();

                if (nativeConnection instanceof RedisAsyncCommands) {
                    RedisAsyncCommands commands = (RedisAsyncCommands) nativeConnection;
                    return (Long) commands.getStatefulConnection().sync().eval(u, ScriptOutputType.INTEGER, keys, values);
                } else if (nativeConnection instanceof RedisAdvancedClusterAsyncCommands) {
                    RedisAdvancedClusterAsyncCommands commands = (RedisAdvancedClusterAsyncCommands) nativeConnection;
                    return (Long) commands.getStatefulConnection().sync().eval(u, ScriptOutputType.INTEGER, keys, values);
                }
                return 0L;
            });
            logger.info("延遲隊列[1]，消息推送成功進入等待隊列({}), topic: {}", result != null && result > 0, e.getTopic());
        } catch (Throwable t) {
            t.printStackTrace();
        }
    }

    private byte[] serialize(String key) {
        RedisSerializer<String> stringRedisSerializer =
                (RedisSerializer<String>) redisTemplate.getKeySerializer();
        //lettuce連接包下序列化鍵值，否則無法用默認的ByteArrayCodec解析
        return stringRedisSerializer.serialize(key);
    }
}

4. 定時任務

/**
 * 分發(fā)任務
 */
@Component
public class DistributeTask {

    private static final String LUA_SCRIPT;
    private Logger logger = LoggerFactory.getLogger(getClass());
    @Autowired
    private StringRedisTemplate redisTemplate;

    static {
        StringBuilder sb = new StringBuilder(128);
        sb.append("local val = redis.call('zrangebyscore', KEYS[1], '-inf', ARGV[1], 'limit', 0, 1)\n");
        sb.append("if(next(val) ~= nil) then\n");
        sb.append("    redis.call('sadd', KEYS[2], ARGV[2])\n");
        sb.append("    redis.call('zremrangebyrank', KEYS[1], 0, #val - 1)\n");
        sb.append("    for i = 1, #val, 100 do\n");
        sb.append("        redis.call('rpush', KEYS[3], unpack(val, i, math.min(i+99, #val)))\n");
        sb.append("    end\n");
        sb.append("    return 1\n");
        sb.append("end\n");
        sb.append("return 0");
        LUA_SCRIPT = sb.toString();
    }

    /**
     * 2秒鐘掃描一次執(zhí)行隊列
     */
    @Scheduled(cron = "0/5 * * * * ?")
    public void scheduledTaskByCorn() {
        try {
            Set<String> members = redisTemplate.opsForSet().members(META_TOPIC_WAIT);
            assert members != null;
            for (String k : members) {
                if (!redisTemplate.hasKey(k)) {
                    // 如果 KEY 不存在元數(shù)據(jù)中刪除
                    redisTemplate.opsForSet().remove(META_TOPIC_WAIT, k);
                    continue;
                }

                String lk = k.replace("delay:wait", "delay:active");
                Object[] keys = new Object[]{serialize(k), serialize(META_TOPIC_ACTIVE), serialize(lk)};
                Object[] values = new Object[]{serialize(String.valueOf(System.currentTimeMillis())), serialize(lk)};
                Long result = redisTemplate.execute((RedisCallback<Long>) connection -> {
                    Object nativeConnection = connection.getNativeConnection();

                    if (nativeConnection instanceof RedisAsyncCommands) {
                        RedisAsyncCommands commands = (RedisAsyncCommands) nativeConnection;
                        return (Long) commands.getStatefulConnection().sync().eval(LUA_SCRIPT, ScriptOutputType.INTEGER, keys, values);
                    } else if (nativeConnection instanceof RedisAdvancedClusterAsyncCommands) {
                        RedisAdvancedClusterAsyncCommands commands = (RedisAdvancedClusterAsyncCommands) nativeConnection;
                        return (Long) commands.getStatefulConnection().sync().eval(LUA_SCRIPT, ScriptOutputType.INTEGER, keys, values);
                    }
                    return 0L;
                });
                logger.info("延遲隊列[2]，消息到期進入執(zhí)行隊列({}): {}", result != null && result > 0, TOPIC_ACTIVE);
            }
        } catch (Throwable t) {
            t.printStackTrace();
        }
    }

    private byte[] serialize(String key) {
        RedisSerializer<String> stringRedisSerializer =
                (RedisSerializer<String>) redisTemplate.getKeySerializer();
        //lettuce連接包下序列化鍵值，否則無法用默認的ByteArrayCodec解析
        return stringRedisSerializer.serialize(key);
    }
}

GitHub 地址

https://github.com/zhengsh/redis-delay-queue