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

 更新时间:2022年09月26日 15:57:07   作者:法力损毁  
这篇文章主要为大家介绍了生产redisson延时队列不消费问题排查解决,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪
(福利推荐:【腾讯云】服务器最新限时优惠活动,云服务器1核2G仅99元/年、2核4G仅768元/3年,立即抢购>>>:9i0i.cn/qcloud

(福利推荐:你还在原价购买阿里云服务器?现在阿里云0.8折限时抢购活动来啦!4核8G企业云服务器仅2998元/3年,立即抢购>>>:9i0i.cn/aliyun

问题描述

项目使用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中获取过期消息如此循环。

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

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

以上就是生产redisson延时队列不消费问题排查解决的详细内容,更多关于排查redisson延时队列不消费的资料请关注程序员之家其它相关文章!

相关文章

  • Redis学习教程之命令的执行过程详解

    Redis学习教程之命令的执行过程详解

    这篇文章主要给大家介绍了关于Redis学习教程之命令的执行过程的相关资料,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧。
    2018-03-03
  • redisson锁tryLock的正确使用方式

    redisson锁tryLock的正确使用方式

    这篇文章主要介绍了redisson锁tryLock的正确使用方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
    2023-12-12
  • 使用Redis获取数据转json,解决动态泛型传参的问题

    使用Redis获取数据转json,解决动态泛型传参的问题

    这篇文章主要介绍了使用Redis获取数据转json,解决动态泛型传参的问题,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2020-07-07
  • redis常用命令小结

    redis常用命令小结

    这篇文章主要介绍了redis的一些常用命令,需要的朋友可以参考下
    2014-06-06
  • Redis哨兵监控的使用

    Redis哨兵监控的使用

    在Redis集群模式中,哨兵模式是一种常用的方案,本文主要介绍了Redis哨兵监控的使用,具有一定的参考价值,感兴趣的可以了解一下
    2023-11-11
  • Redis Stat的安装指南

    Redis Stat的安装指南

    这篇文章主要介绍了Redis Stat的安装指南的相关资料,需要的朋友可以参考下
    2016-04-04
  • 使用Redis实现向量相似度搜索

    使用Redis实现向量相似度搜索

    在自然语言处理领域,有一个常见且重要的任务就是文本相似度搜索,所以本文为大家介绍一下如何利用Redis实现向量相似度搜索,解决文本、图像和音频之间的相似度匹配问题,需要的可以了解下
    2023-07-07
  • 利用redis实现分布式锁,快速解决高并发时的线程安全问题

    利用redis实现分布式锁,快速解决高并发时的线程安全问题

    这篇文章主要介绍了利用redis实现分布式锁,快速解决高并发时的线程安全问题,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2021-01-01
  • Redis出现(error)NOAUTH?Authentication?required.报错的解决办法(秒懂!)

    Redis出现(error)NOAUTH?Authentication?required.报错的解决办法(秒懂!)

    这篇文章主要给大家介绍了关于Redis出现(error)NOAUTH?Authentication?required.报错的解决办法,对于 这个错误这通常是因为Redis服务器需要密码进行身份验证,但客户端没有提供正确的身份验证信息导致的,需要的朋友可以参考下
    2024-03-03
  • redis调用二维码时的不断刷新排查分析

    redis调用二维码时的不断刷新排查分析

    这篇文章主要为大家介绍了redis调用二维码时不断刷新排查分析,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步早日升职加薪
    2022-04-04

最新评论

?


http://www.vxiaotou.com