排查redisson延时队列不消费

时间:2022-09-29 法力损毁人气:0

问题描述

项目使用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)方法

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

这里可以看到添加了两个listener，onSubcribe，onMessage，当订阅到消息时执行onSubcribe中的pushTash，当redis有新的消息通知，就会触发scheduleTask(...)方法，startTime为上述中publish通知的元素过期时间

继续进入pushTask方法

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

继续进入scheduleTask()方法

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

流程总结

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

发送端只是往zset，list，添加数据，并且发布一条订阅消息

消费端收到订阅消息后会查询zset中的过期消息，并放入阻塞队列供消费端take消息，并且获取zset第一个消息，启动一个延时任务，到期后继续从zset中获取过期消息如此循环。

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

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

加载全部内容