?背景

前段時(shí)間有個(gè)小項(xiàng)目需要使用延遲任務(wù)，談到延遲任務(wù)，我腦子第一時(shí)間一閃而過(guò)的就是使用消息隊(duì)列來(lái)做，比如RabbitMQ的死信隊(duì)列又或者RocketMQ的延遲隊(duì)列，但是奈何這是一個(gè)小項(xiàng)目，并沒(méi)有引入MQ，我也不太想因?yàn)橐粋€(gè)延遲任務(wù)就引入MQ，增加系統(tǒng)復(fù)雜度，所以這個(gè)方案直接就被pass了。

雖然基于MQ這個(gè)方式走不通了，但是這個(gè)項(xiàng)目中使用到Redis，所以我就想是否能夠使用Redis來(lái)代替MQ實(shí)現(xiàn)延遲隊(duì)列的功能，于是我就查了一下有沒(méi)有現(xiàn)成可用的方案，別說(shuō)，還真給我查到了兩種方案，并且我還仔細(xì)研究對(duì)比了這兩個(gè)方案，發(fā)現(xiàn)要想很好的實(shí)現(xiàn)延遲隊(duì)列，并不簡(jiǎn)單。

監(jiān)聽(tīng)過(guò)期key

基于監(jiān)聽(tīng)過(guò)期key的方式來(lái)實(shí)現(xiàn)延遲隊(duì)列是我查到的第一個(gè)方案，為了弄懂這個(gè)方案實(shí)現(xiàn)的細(xì)節(jié)，我還特地去扒了扒官網(wǎng)，還真有所收獲

1、Redis發(fā)布訂閱模式

一談到發(fā)布訂閱模式，其實(shí)一想到的就是MQ，只不過(guò)Redis也實(shí)現(xiàn)了一套，并且跟MQ賊像，如圖：

圖中的channel的概念跟MQ中的topic的概念差不多，你可以把channel理解成MQ中的topic。

生產(chǎn)者在消息發(fā)送時(shí)需要到指定發(fā)送到哪個(gè)channel上，消費(fèi)者訂閱這個(gè)channel就能獲取到消息。

2、keyspace notifications

在Redis中，有很多默認(rèn)的channel，只不過(guò)向這些channel發(fā)送消息的生產(chǎn)者不是我們寫的代碼，而是Redis本身。當(dāng)消費(fèi)者監(jiān)聽(tīng)這些channel時(shí)，就可以感知到Redis中數(shù)據(jù)的變化。

這個(gè)功能Redis官方稱為keyspace notifications，字面意思就是鍵空間通知。

這些默認(rèn)的channel被分為兩類：

• 以__keyspace@<db>__:為前綴，后面跟的是key的名稱，表示監(jiān)聽(tīng)跟這個(gè)key有關(guān)的事件。舉個(gè)例子，現(xiàn)在有個(gè)消費(fèi)者監(jiān)聽(tīng)了__keyspace@0__:sanyou這個(gè)channel，sanyou就是Redis中的一個(gè)普通key，那么當(dāng)sanyou這個(gè)key被刪除或者發(fā)生了其它事件，那么消費(fèi)者就會(huì)收到sanyou這個(gè)key刪除或者其它事件的消息
• 以__keyevent@<db>__:為前綴，后面跟的是消息事件類型，表示監(jiān)聽(tīng)某個(gè)事件同樣舉個(gè)例子，現(xiàn)在有個(gè)消費(fèi)者監(jiān)聽(tīng)了__keyevent@0__:expired這個(gè)channel，代表了監(jiān)聽(tīng)key的過(guò)期事件。那么當(dāng)某個(gè)Redis的key過(guò)期了（expired），那么消費(fèi)者就能收到這個(gè)key過(guò)期的消息。如果把expired換成del，那么監(jiān)聽(tīng)的就是刪除事件。具體支持哪些事件，可從官網(wǎng)查。

上述db是指具體的數(shù)據(jù)庫(kù)，Redis不是默認(rèn)分為16個(gè)庫(kù)么，序號(hào)從0-15，所以db就是0到15的數(shù)字，示例中的0就是指0對(duì)應(yīng)的數(shù)據(jù)庫(kù)。

3、延遲隊(duì)列實(shí)現(xiàn)原理

通過(guò)對(duì)上面的兩個(gè)概念了解之后，應(yīng)該就對(duì)監(jiān)聽(tīng)過(guò)期key的實(shí)現(xiàn)原理一目了然了，其實(shí)就是當(dāng)這個(gè)key過(guò)期之后，Redis會(huì)發(fā)布一個(gè)key過(guò)期的事件到__keyevent@<db>__:expired這個(gè)channel，只要我們的服務(wù)監(jiān)聽(tīng)這個(gè)channel，那么就能知道過(guò)期的Key，從而就算實(shí)現(xiàn)了延遲隊(duì)列功能。

所以這種方式實(shí)現(xiàn)延遲隊(duì)列就只需要兩步：

• 發(fā)送延遲任務(wù)，key是延遲消息本身，過(guò)期時(shí)間就是延遲時(shí)間
• 監(jiān)聽(tīng)__keyevent@<db>__:expired這個(gè)channel，處理延遲任務(wù)

4、demo

好了，基本概念和核心原理都說(shuō)完了之后，又到了show me the code環(huán)節(jié)。

好巧不巧，Spring已經(jīng)實(shí)現(xiàn)了監(jiān)聽(tīng)__keyevent@*__:expired?這個(gè)channel這個(gè)功能，__keyevent@*__:expired?中的*代表通配符的意思，監(jiān)聽(tīng)所有的數(shù)據(jù)庫(kù)。

所以demo寫起來(lái)就很簡(jiǎn)單了，只需3步即可

引入pom

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
    <version>2.2.5.RELEASE</version>
</dependency>

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
    <version>2.2.5.RELEASE</version>
</dependency>

配置類

@Configuration
public class RedisConfiguration {

    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory) {
        RedisMessageListenerContainer redisMessageListenerContainer = new RedisMessageListenerContainer();
        redisMessageListenerContainer.setConnectionFactory(connectionFactory);
        return redisMessageListenerContainer;
    }

    @Bean
    public KeyExpirationEventMessageListener redisKeyExpirationListener(RedisMessageListenerContainer redisMessageListenerContainer) {
        return new KeyExpirationEventMessageListener(redisMessageListenerContainer);
    }

}

KeyExpirationEventMessageListener實(shí)現(xiàn)了對(duì)__keyevent@*__:expiredchannel的監(jiān)聽(tīng)

當(dāng)KeyExpirationEventMessageListener收到Redis發(fā)布的過(guò)期Key的消息的時(shí)候，會(huì)發(fā)布RedisKeyExpiredEvent事件

所以我們只需要監(jiān)聽(tīng)RedisKeyExpiredEvent事件就可以拿到過(guò)期消息的Key，也就是延遲消息。

對(duì)RedisKeyExpiredEvent事件的監(jiān)聽(tīng)實(shí)現(xiàn)MyRedisKeyExpiredEventListener

@Component
public class MyRedisKeyExpiredEventListener implements ApplicationListener<RedisKeyExpiredEvent> {

    @Override
    public void onApplicationEvent(RedisKeyExpiredEvent event) {
        byte[] body = event.getSource();
        System.out.println("獲取到延遲消息：" + new String(body));
    }

}

整個(gè)工程目錄也簡(jiǎn)單

代碼寫好，啟動(dòng)應(yīng)用

之后我直接通過(guò)Redis命令設(shè)置消息，就沒(méi)通過(guò)代碼發(fā)送消息了，消息的key為sanyou，值為task，值不重要，過(guò)期時(shí)間為5s

set sanyou task 

expire sanyou 5

如果上面都理論都正確，不出意外的話，5s后MyRedisKeyExpiredEventListener應(yīng)該可以監(jiān)聽(tīng)到sanyou這個(gè)key過(guò)期的消息，也就相當(dāng)于拿到了延遲任務(wù)，控制臺(tái)會(huì)打印出獲取到延遲消息：sanyou。

于是我滿懷希望，靜靜地等待了5s。。

5、4、3、2、1，時(shí)間一到，我查看控制臺(tái)，但是控制臺(tái)并沒(méi)有按照預(yù)期打印出上面那句話。

為什么會(huì)沒(méi)打印出？難道是代碼寫錯(cuò)了？正當(dāng)我準(zhǔn)備檢查代碼的時(shí)候，官網(wǎng)的一段話道出了真實(shí)原因。

我給大家翻譯一下上面這段話講的內(nèi)容。

上面這段話主要討論的是key過(guò)期事件的時(shí)效性問(wèn)題，首先提到了Redis過(guò)期key的兩種清除策略，就是面試八股文常背的兩種：

• 惰性清除。當(dāng)這個(gè)key過(guò)期之后，訪問(wèn)時(shí)，這個(gè)Key才會(huì)被清除
• 定時(shí)清除。后臺(tái)會(huì)定期檢查一部分key，如果有key過(guò)期了，就會(huì)被清除

再后面那段話是核心，意思是說(shuō)，key的過(guò)期事件發(fā)布時(shí)機(jī)并不是當(dāng)這個(gè)key的過(guò)期時(shí)間到了之后就發(fā)布，而是這個(gè)key在Redis中被清理之后，也就是真正被刪除之后才會(huì)發(fā)布。

到這我終于明白了，上面的例子中即使我設(shè)置了5s的過(guò)期時(shí)間，但是當(dāng)5s過(guò)去之后，只要兩種清除策略都不滿足，沒(méi)人訪問(wèn)sanyou這個(gè)key，后臺(tái)的定時(shí)清理的任務(wù)也沒(méi)掃描到sanyou這個(gè)key，那么就不會(huì)發(fā)布key過(guò)期的事件，自然而然也就監(jiān)聽(tīng)不到了。

至于后臺(tái)的定時(shí)清理的任務(wù)什么時(shí)候能掃到，這個(gè)沒(méi)有固定時(shí)間，可能一到過(guò)期時(shí)間就被掃到，也可能等一定時(shí)間才會(huì)被掃到，這就可能會(huì)造成了客戶端從發(fā)布到監(jiān)聽(tīng)到的消息時(shí)間差會(huì)大于等于過(guò)期時(shí)間，從而造成一定時(shí)間消息的延遲，這就著實(shí)有點(diǎn)坑了。。

5、坑

除了上面測(cè)試demo的時(shí)候遇到的坑之外，在我深入研究之后，還發(fā)現(xiàn)了一些更離譜的坑。

丟消息太頻繁

Redis的丟消息跟MQ不一樣，因?yàn)镸Q都會(huì)有消息的持久化機(jī)制，可能只有當(dāng)機(jī)器宕機(jī)了，才會(huì)丟點(diǎn)消息，但是Redis丟消息就很離譜，比如說(shuō)你的服務(wù)在重啟的時(shí)候就消息會(huì)丟消息。

Redis實(shí)現(xiàn)的發(fā)布訂閱模式，消息是沒(méi)有持久化機(jī)制，當(dāng)消息發(fā)布到某個(gè)channel之后，如果沒(méi)有客戶端訂閱這個(gè)channel，那么這個(gè)消息就丟了，并不會(huì)像MQ一樣進(jìn)行持久化，等有消費(fèi)者訂閱的時(shí)候再給消費(fèi)者消費(fèi)。

所以說(shuō)，假設(shè)服務(wù)重啟期間，某個(gè)生產(chǎn)者或者是Redis本身發(fā)布了一條消息到某個(gè)channel，由于服務(wù)重啟，沒(méi)有監(jiān)聽(tīng)這個(gè)channel，那么這個(gè)消息自然就丟了。

消息消費(fèi)只有廣播模式

Redis的發(fā)布訂閱模式消息消費(fèi)只有廣播模式一種。

所謂的廣播模式就是多個(gè)消費(fèi)者訂閱同一個(gè)channel，那么每個(gè)消費(fèi)者都能消費(fèi)到發(fā)布到這個(gè)channel的所有消息。

如圖，生產(chǎn)者發(fā)布了一條消息，內(nèi)容為sanyou，那么兩個(gè)消費(fèi)者都可以同時(shí)收到sanyou這條消息。

所以，如果通過(guò)監(jiān)聽(tīng)channel來(lái)獲取延遲任務(wù)，那么一旦服務(wù)實(shí)例有多個(gè)的話，還得保證消息不能重復(fù)處理，額外地增加了代碼開(kāi)發(fā)量。

接收到所有key的某個(gè)事件

這個(gè)不屬于Redis發(fā)布訂閱模式的問(wèn)題，而是Redis本身事件通知的問(wèn)題。

當(dāng)消費(fèi)者監(jiān)聽(tīng)了以__keyevent@<db>__:開(kāi)頭的消息，那么會(huì)導(dǎo)致所有的key發(fā)生了事件都會(huì)被通知給消費(fèi)者。

舉個(gè)例子，某個(gè)消費(fèi)者監(jiān)聽(tīng)了__keyevent@*__:expired這個(gè)channel，那么只要key過(guò)期了，不管這個(gè)key是張三還會(huì)李四，消費(fèi)者都能收到。

所以如果你只想消費(fèi)某一類消息的key，那么還得自行加一些標(biāo)記，比如消息的key加個(gè)前綴，消費(fèi)的時(shí)候判斷一下帶前綴的key就是需要消費(fèi)的任務(wù)。

所以，綜上能夠得出一個(gè)非常重要的結(jié)論，那就是監(jiān)聽(tīng)Redis過(guò)期Key這種方式實(shí)現(xiàn)延遲隊(duì)列，不穩(wěn)定，坑賊多！

那有沒(méi)有比較靠譜的延遲隊(duì)列的實(shí)現(xiàn)方案呢？這就不得不提到我研究的第二種方案了。

Redisson實(shí)現(xiàn)延遲隊(duì)列

Redisson他是Redis的兒子（Redis son），基于Redis實(shí)現(xiàn)了非常多的功能，其中最常使用的就是Redis分布式鎖的實(shí)現(xiàn)，但是除了實(shí)現(xiàn)Redis分布式鎖之外，它還實(shí)現(xiàn)了延遲隊(duì)列的功能。

先來(lái)個(gè)demo，后面再來(lái)說(shuō)說(shuō)這種實(shí)現(xiàn)的原理。

1、demo

引入pom

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.13.1</version>
</dependency>

封裝了一個(gè)RedissonDelayQueue類

@Component
@Slf4j
public class RedissonDelayQueue {

    private RedissonClient redissonClient;

    private RDelayedQueue<String> delayQueue;
    private RBlockingQueue<String> blockingQueue;

    @PostConstruct
    public void init() {
        initDelayQueue();
        startDelayQueueConsumer();
    }

    private void initDelayQueue() {
        Config config = new Config();
        SingleServerConfig serverConfig = config.useSingleServer();
        serverConfig.setAddress("redis://localhost:6379");
        redissonClient = Redisson.create(config);

        blockingQueue = redissonClient.getBlockingQueue("SANYOU");
        delayQueue = redissonClient.getDelayedQueue(blockingQueue);
    }

    private void startDelayQueueConsumer() {
        new Thread(() -> {
            while (true) {
                try {
                    String task = blockingQueue.take();
                    log.info("接收到延遲任務(wù):{}", task);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }, "SANYOU-Consumer").start();
    }

    public void offerTask(String task, long seconds) {
        log.info("添加延遲任務(wù):{} 延遲時(shí)間:{}s", task, seconds);
        delayQueue.offer(task, seconds, TimeUnit.SECONDS);
    }

}

這個(gè)類在創(chuàng)建的時(shí)候會(huì)去初始化延遲隊(duì)列，創(chuàng)建一個(gè)RedissonClient對(duì)象，之后通過(guò)RedissonClient對(duì)象獲取到RDelayedQueue和RBlockingQueue對(duì)象，傳入的隊(duì)列名字叫SANYOU，這個(gè)名字無(wú)所謂。

當(dāng)延遲隊(duì)列創(chuàng)建之后，會(huì)開(kāi)啟一個(gè)延遲任務(wù)的消費(fèi)線程，這個(gè)線程會(huì)一直從RBlockingQueue中通過(guò)take方法阻塞獲取延遲任務(wù)。

添加任務(wù)的時(shí)候是通過(guò)RDelayedQueue的offer方法添加的。

controller類，通過(guò)接口添加任務(wù)，延遲時(shí)間為5s

@RestController
public class RedissonDelayQueueController {

    @Resource
    private RedissonDelayQueue redissonDelayQueue;

    @GetMapping("/add")
    public void addTask(@RequestParam("task") String task) {
        redissonDelayQueue.offerTask(task, 5);
    }

}

啟動(dòng)項(xiàng)目，在瀏覽器輸入如下連接，添加任務(wù)

http://localhost:8080/add?task=sanyou

靜靜等待5s，成功獲取到任務(wù)。

2、實(shí)現(xiàn)原理

如下圖就是上面demo中，一個(gè)延遲隊(duì)列會(huì)在Redis內(nèi)部使用到的channel和數(shù)據(jù)類型

SANYOU前面的前綴都是固定的，Redisson創(chuàng)建的時(shí)候會(huì)拼上前綴。

• redisson_delay_queue_timeout:SANYOU，sorted set數(shù)據(jù)類型，存放所有延遲任務(wù)，按照延遲任務(wù)的到期時(shí)間戳（提交任務(wù)時(shí)的時(shí)間戳 + 延遲時(shí)間）來(lái)排序的，所以列表的最前面的第一個(gè)元素就是整個(gè)延遲隊(duì)列中最早要被執(zhí)行的任務(wù)，這個(gè)概念很重要
• redisson_delay_queue:SANYOU，list數(shù)據(jù)類型，也是存放所有的任務(wù)，但是研究下來(lái)發(fā)現(xiàn)好像沒(méi)什么用。。
• SANYOU，list數(shù)據(jù)類型，被稱為目標(biāo)隊(duì)列，這個(gè)里面存放的任務(wù)都是已經(jīng)到了延遲時(shí)間的，可以被消費(fèi)者獲取的任務(wù)，所以上面demo中的RBlockingQueue的take方法是從這個(gè)目標(biāo)隊(duì)列中獲取到任務(wù)的
• redisson_delay_queue_channel:SANYOU，是一個(gè)channel，用來(lái)通知客戶端開(kāi)啟一個(gè)延遲任務(wù)

有了這些概念之后，再來(lái)看看整體的運(yùn)行原理圖

• 生產(chǎn)者在提交任務(wù)的時(shí)候?qū)⑷蝿?wù)放到redisson_delay_queue_timeout:SANYOU中，分?jǐn)?shù)就是提交任務(wù)的時(shí)間戳+延遲時(shí)間，就是延遲任務(wù)的到期時(shí)間戳
• 客戶端會(huì)有一個(gè)延遲任務(wù)，為了區(qū)分，后面我都說(shuō)是客戶端延遲任務(wù)。這個(gè)延遲任務(wù)會(huì)向Redis Server發(fā)送一段lua腳本，Redis執(zhí)行l(wèi)ua腳本中的命令，并且是原子性的

這段lua腳本主要干了兩件事：

• 將到了延遲時(shí)間的任務(wù)從redisson_delay_queue_timeout:SANYOU中移除，存到SANYOU這個(gè)目標(biāo)隊(duì)列
• 獲取到redisson_delay_queue_timeout:SANYOU中目前最早到過(guò)期時(shí)間的延遲任務(wù)的到期時(shí)間戳，然后發(fā)布到redisson_delay_queue_channel:SANYOU這個(gè)channel中

當(dāng)客戶端監(jiān)聽(tīng)到redisson_delay_queue_channel:SANYOU?這個(gè)channel的消息時(shí)，會(huì)再次提交一個(gè)客戶端延遲任務(wù)，延遲時(shí)間就是消息（最早到過(guò)期時(shí)間的延遲任務(wù)的到期時(shí)間戳）- 當(dāng)前時(shí)間戳，這個(gè)時(shí)間其實(shí)也就是redisson_delay_queue_channel:SANYOU中最早到過(guò)期時(shí)間的任務(wù)還剩余的延遲時(shí)間。

此處可以等待10s，好好想想。。

這樣，一旦時(shí)間來(lái)到了上面說(shuō)的最早到過(guò)期時(shí)間任務(wù)的到期時(shí)間戳，redisson_delay_queue_timeout:SANYOU?中上面說(shuō)的最早到過(guò)期時(shí)間的任務(wù)已經(jīng)到期了，客戶端的延遲任務(wù)也同時(shí)到期，于是開(kāi)始執(zhí)行l(wèi)ua腳本操作，及時(shí)將到了延遲時(shí)間的任務(wù)放到目標(biāo)隊(duì)列中。然后再次發(fā)布剩余的延遲任務(wù)中最早到期的任務(wù)到期時(shí)間戳到channe中，如此循環(huán)往復(fù)，一直運(yùn)行下去，保證redisson_delay_queue_timeout:SANYOU中到期的數(shù)據(jù)能及時(shí)放到目標(biāo)隊(duì)列中。

所以，上述說(shuō)了一大堆的主要的作用就是保證到了延遲時(shí)間的任務(wù)能夠及時(shí)被放到目標(biāo)隊(duì)列。

這里再補(bǔ)充兩個(gè)特殊情況，圖中沒(méi)有畫出：

第一個(gè)就是如果redisson_delay_queue_timeout:SANYOU是新添加的任務(wù)（隊(duì)列之前有或者沒(méi)有任務(wù)）是隊(duì)列中最早需要被執(zhí)行的，也會(huì)發(fā)布消息到channel，之后就按時(shí)上面說(shuō)的流程走了。

添加任務(wù)代碼如下，也是通過(guò)lua腳本來(lái)的

第二種特殊情況就是項(xiàng)目啟動(dòng)的時(shí)候會(huì)執(zhí)行一次客戶端延遲任務(wù)。項(xiàng)目在重啟時(shí)，由于沒(méi)有客戶端延遲任務(wù)的執(zhí)行，可能會(huì)出現(xiàn)redisson_delay_queue_timeout:SANYOU隊(duì)列中有到期但是沒(méi)有被放到目標(biāo)隊(duì)列的可能，重啟就執(zhí)行一次就是為了保證到期的數(shù)據(jù)能被及時(shí)放到目標(biāo)隊(duì)列中。

3、與第一種方案比較

現(xiàn)在來(lái)比較一下第一種方案和Redisson的這種方案，看看有沒(méi)有第一種方案的那些坑。

第一個(gè)任務(wù)延遲的問(wèn)題，Redisson方案理論上是沒(méi)有延遲的，但是當(dāng)消息數(shù)量增加，消費(fèi)者消費(fèi)緩慢這個(gè)情況下可能會(huì)導(dǎo)致延遲任務(wù)消費(fèi)的延遲。

第二個(gè)丟消息的問(wèn)題，Redisson方案很大程度上減輕了丟消息的可能性，因?yàn)樗械娜蝿?wù)都是存在list和sorted set兩種數(shù)據(jù)類型中，Redis有持久化機(jī)制，就算Redis宕機(jī)了，也就可能會(huì)丟一點(diǎn)點(diǎn)數(shù)據(jù)。

第三個(gè)廣播消費(fèi)任務(wù)的問(wèn)題，這個(gè)是不會(huì)出現(xiàn)的，因?yàn)槊總€(gè)客戶端都是從同一個(gè)目標(biāo)隊(duì)列中獲取任務(wù)的。

第四個(gè)問(wèn)題是Redis內(nèi)部channel發(fā)布事件的問(wèn)題，跟這種方案不沾邊，就更不可能存在了。

所以，通過(guò)上面的對(duì)比可以看出，Redisson這種實(shí)現(xiàn)方案就顯得更加的靠譜了。