广告位联系
返回顶部
分享到

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

Redis 来源:互联网 作者:佚名 发布时间:2022-09-28 17:24:27 人浏览
摘要

问题描述 项目使用redisson延时队列功能,实现直播的开播提醒,突然有一天业务爆出问题,未触发开播提醒。 初步排查 首先通过查询生产日志,发送端日志存在,没有消费日志,猜测

问题描述

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

初步排查

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

排查过程

发送端代码

1

2

3

4

5

6

7

8

9

10

11

12

13

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);

    }

}

消费端代码

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

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命令,订阅延时队列,这就导致在服务重启前发送的消息到时间后无法消费。

解决方案

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

1

2

//订阅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中获取过期消息如此循环。


版权声明 : 本文内容来源于互联网或用户自行发布贡献,该文观点仅代表原作者本人。本站仅提供信息存储空间服务和不拥有所有权,不承担相关法律责任。如发现本站有涉嫌抄袭侵权, 违法违规的内容, 请发送邮件至2530232025#qq.cn(#换@)举报,一经查实,本站将立刻删除。
原文链接 : https://juejin.cn/post/7147507349442265101
相关文章
  • spring boot集成redis基础入门实例介绍
    redis 支持持久化数据,不仅支持key-value类型的数据,还拥有list,set,zset,hash等数据结构的存储。 可以进行master-slave模式的数据备份 更多
  • redis批量操作pipeline管道操作方法

    redis批量操作pipeline管道操作方法
    redis | pipeline(管道) 背景 Redis是一种基于客户端-服务端模型以及请求/响应的TCP服务。这意味着通常情况下一个请求会遵循以下步骤: 客户
  • springboot整合使用云服务器上的Redis方法

    springboot整合使用云服务器上的Redis方法
    一、前提条件 修改redis.conf配置文件 1、protected-mode yes(默认的) 修改成 protected-mode no,解除保护模式 2、注释掉绑定ip ,绑定ip的话,使得
  • 阿里云服务器部署Redis并整合Spring Boot的介绍

    阿里云服务器部署Redis并整合Spring Boot的介绍
    一、什么是Redis redis是一个key-value存储系统。和Memcached类似,它支持存储的value类型相对更多,包括string(字符串)、list(链表)、set(集合)、zs
  • 生产redisson延时队列不消费问题排查解决

    生产redisson延时队列不消费问题排查解决
    问题描述 项目使用redisson延时队列功能,实现直播的开播提醒,突然有一天业务爆出问题,未触发开播提醒。 初步排查 首先通过查询生产日
  • Redis主从复制分步讲解使用

    Redis主从复制分步讲解使用
    主服务器(master)启用二进制日志 选择一个唯一的server-id 创建具有复制权限的用户 从服务器(slave)启用中继日志, 选择一个唯一的serv
  • Redis中HyperLogLog的使用介绍
    HyperLogLog,基数统计; 那什么是基数? 比如有两个数组 数组A = [1,2,3,4,5]; 数组B = [3,4,5,6,7]; 这时候基数就是[1,2,3,4,5,6,7],总共有7个数; 就是
  • Redis中的持久化介绍

    Redis中的持久化介绍
    1. 前言 为什么要进行持久化?:持久化功能有效地避免因进程退出造成的数据丢失问题,当下次重启时利用之前持久化的文件即可实现数据
  • Redis源码设计剖析之事件处理示例介绍

    Redis源码设计剖析之事件处理示例介绍
    1. Redis事件介绍 Redis服务器是一个 事件驱动程序 ,所谓事件驱动就是输入一条命令并且按下回车,然后消息被组装成 Redis 协议的格式发送给
  • Mysql应用安装后找不到my.ini文件的解决过程

    Mysql应用安装后找不到my.ini文件的解决过程
    一、背景 我在两台电脑上安装了MySQL Server 8.0,准备继续做主从配置,这时候就需要用到my.ini文件进行配置,但是我找不到my.ini文件。 我的
  • 本站所有内容来源于互联网或用户自行发布,本站仅提供信息存储空间服务,不拥有版权,不承担法律责任。如有侵犯您的权益,请您联系站长处理!
  • Copyright © 2017-2022 F11.CN All Rights Reserved. F11站长开发者网 版权所有 | 苏ICP备2022031554号-1 | 51LA统计