一次生产redisson 延时队列不消费问题排查

# 问题描述

项目使用 redisson 延时队列功能，实现直播的开播提醒，突然有一天业务爆出问题，未触发开播提醒。

# 初步排查

首先通过查询生产日志，发送端日志存在，没有消费日志，猜测消费端没有消费到延时消息，，在 dba 的协助下查询 redis 队列，消息也确实存在，但已经过了过期时间，由此证明 redisson 消费者出现问题。通过服务日志发现在最后一次设置自定义推送任务是在一次服务发布之前，服务发布后，之前设置的自定义推送消息均没有被客户端消费，由此猜想是由发布服务导致消费端失效。

# 排查过程

发送端代码

```java
public  void produce(String delayQueue, T t, long delay, TimeUnit timeUnit) {
    try {
        log.info("delay msg,delayQueue:{},key:{},delay:{}", delayQueue, t, delay);
        if (delay < 0) {
            delay = 0;
        }
        RBlockingQueue blockingFairQueue = redissonClient.getBlockingQueue(delayQueue);
        RDelayedQueue delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);
        delayedQueue.offer(t, delay, timeUnit);
    }catch (Exception e){
        log.error("添加延时任务队列失败",e);
    }
}
````

消费端代码

```java
public class DelayTaskHandler implements Runnable {
    @Override
    public void run() {
        RBlockingQueue blockingFairQueue = redissonClient.getBlockingQueue(delayQueue);
        while (true) {
            try {
                T value = blockingFairQueue.take();
                log.info("delay queue {},延时任务开始执行,value - {} , timeStamp - {} , threadName - {}", delayQueue, value, System.currentTimeMillis(), Thread.currentThread().getName());
                consumer.accept(value);
            } catch (Exception e) {
                log.error("延时任务执行失败,", e);
            }
        }
    }
}
```

**因为 redisson 延时队列是基于 redis 实现的，所以从 redis 执行命令开始入手排查**

> 1.打开 redis 监控，启动服务,发现 redis 首先执行了 blpop 命令，阻塞等待{cl-live-admin:notice_delay_queue} 队列消息
> ![image (1).png](/media/202307/2023-07-06_1607320.8506564435197156.png)

> 2.提交一个延时任务后，观察 redis 命令
> ![image (2).png](/media/202307/2023-07-06_1607320.710497247352768.png)
> 此时发现 redis 首先执行了一个 SUBSCRIBE 命令，订阅了一个队列，然后执行了一段 lua 脚本，主要包括以下命令：

- zrangebyscore:获取 zset 中 score 在 0 至当前时间戳范围内的前一百条数据 如果获取到数据则循环执行 rpush，lrem，zrem 命令
- zrange:取 zset 中第一条数据
- zadd:向 zset 中添加一条数据，score 为时间戳
- rpush：向 list 右边 push 一条数据
- publish：如果添加的消息在顶部，则发布一条订阅消息

> 3.消费一条消息
> ![image (3).png](/media/202307/2023-07-06_1607320.28217172136451185.png)
> 同样消费的时候也是提交了一条 lua 脚本，主要执行了以下命令 可以看到和发送端命令相似

- zrangebyscore：获取 zset 中 score 在 0 至当前时间戳范围内的前一百条数据
- rpush：向 list 右边 push 一条数据
- lrem：删除一条数据
- zrem：删除 zeset 中的数据
- zrange：获取第一条数据
- BLPOP：阻塞等待队列消息

通过以上 redis 命令的执行可以发现一个命令 SUBCRIBE 用于订阅 redis 的一个队列，而这个命令只在发送消息的时候执行了，在消费的时候没有执行。从而验证了当服务重启后如果没有新的消息发送，那么客户端就不会发送 SUBCRIBE 命令，订阅延时队列，这就导致在服务重启前发送的消息到时间后无法消费。

# 解决方案

在消费端启动的时候添加一行代码用于订阅延时队列

```java
//订阅 redis 队列
 redissonClient.getDelayedQueue(blockingFairQueue);
```

**那么为什么没有订阅就消费不到消息了呢？带着疑问继续深入理解 redisson 的实现**

# redisson 延时队列原理

![image (4).png](/media/202307/2023-07-06_1607320.9068129330953285.png)

> 首先回到消费端代码

![image (5).png](/media/202307/2023-07-06_1607320.20098528195970444.png)

> 在我们没有发送订阅命令的时候，客户端只是在阻塞等待一个指定队列的消息，那么这个队列的消息是谁放进去的呢？
> 带着疑问我们再看发送端代码

![image (6).png](/media/202307/2023-07-06_1607320.7713453052373151.png)

> 直接进入 delayedQueue.offer()方法内部

![image (7).png](/media/202307/2023-07-06_1607320.9982939324016893.png)

> 可以看到发送端是提交了一个 lua 脚本主要执行了 zadd，rpush，publish 命令，这里我们需要注意 publish 命令，在 redis 中 pub/sub 是对应的，当有 publish 的时候，那么 subcribe 端会收到该订阅消息。

> 那么是谁收到了订阅的消息，收到消息后又做了什么呢，回到 redissonClient.getDelayedQueue(blockingFairQueue)代码中

![image (8).png](/media/202307/2023-07-06_1607320.8353184486631285.png)

> 继续进入 new RedissonDelayedQueue()

![image (9).png](/media/202307/2023-07-06_1607320.942095711522288.png)

> 可以看到这里创建了一个 QueueTransferTask，实现了 pushTaskAsync()方法，具体内容是一个 lua 脚本，首先执行 zrangebyscore 获取过期的前一百条数据，循环调用 rpush，lrem，zrem，注意这里 rpush 的队列为我们指定的延时队列，也就是 consumer 端 take 的队列。至此明白了消费端的消息是方法 pushTaskAsync()执行后放入的。那么什么时候执行这个方法呢。

> 进入 queueTransferService.schedule(queueName, task)方法

![image (10).png](/media/202307/2023-07-06_1607320.11837347700467638.png)

> 这里会执行 start 方法，继续跟进

![image (11).png](/media/202307/2023-07-06_1607320.10534641038863124.png)
这里可以看到添加了两个 listener，onSubcribe，onMessage，当订阅到消息时执行 onSubcribe 中的 pushTash，当 redis 有新的消息通知，就会触发 scheduleTask(...)方法，startTime 为上述中 publish 通知的元素过期时间

> 继续进入 pushTask 方法

![image (12).png](/media/202307/2023-07-06_1607320.40388323324164144.png)

> 这里可以看到一个熟悉的方法 pushTaskAsync()，也就是前边的一段 lua 脚本，用于将过期的消息放入阻塞队列，并返回排在第一个的消息执行 scheduleTask()

> 继续进入 scheduleTask()方法

![image (13).png](/media/202307/2023-07-06_1607320.530879429515256.png)

> 如果时间差小于 10 毫秒则执行 pushTask 方法，如果大于 10 毫秒则启动一个延时任务，到时间后执行 pushTask 方法。pushTask 与 scheduleTask 互相调用循环往复

# 至此源码分析完毕，整个流程总结如下：

> 1. 发送端只是往 zset，list，添加数据，并且发布一条订阅消息
> 2. 消费端收到订阅消息后会查询 zset 中的过期消息，并放入阻塞队列供消费端 take 消息，并且获取 zset 第一个消息，启动一个延时任务，到期后继续从 zset 中获取过期消息如此循环。

此时就回答了上边的问题 **那么为什么没有订阅就消费不到消息了呢？**

如果没有订阅的话消费端就收不到订阅消息，也就不会去获取过期时间放入阻塞队列进行循环。