您当前所在位置：首页 > 数据库 > Redis

Redis

生产redisson延时队列不消费问题排查解决

[db:来源] 法力损毁2022-09-26Redis3805

这篇文章主要为大家介绍了生产redisson延时队列不消费问题排查解决，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪

问题描述

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

初步排查

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

排查过程

发送端代码

public <T> 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<T> blockingFairQueue = redissonClient.getBlockingQueue(delayQueue);
        RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);
        delayedQueue.offer(t, delay, timeUnit);
    }catch (Exception e){
        log.error("添加延时任务队列失败",e);
    }
}

消费端代码

public class DelayTaskHandler implements Runnable {
    @Override
    public void run() {
        RBlockingQueue<T> 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} 队列消息

2.提交一个延时任务后，观察redis命令

此时发现redis首先执行了一个SUBSCRIBE命令，订阅了一个队列，然后执行了一段lua脚本，主要包括以下命令：

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

3.消费一条消息

同样消费的时候也是提交了一条lua脚本，主要执行了以下命令可以看到和发送端命令相似

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

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

解决方案

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

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

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

redisson 延时队列原理

首先回到消费端代码

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

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

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

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

继续进入 new RedissonDelayedQueue()

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

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