devbox
秋刀鱼在做梦
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

springboot&redisson实现延时队列_springboot redission延迟队列

作者：秋刀鱼在做梦 | 2024-07-14 04:42:37

赞

踩

springboot redission延迟队列

Redisson实现延时队列

版本说明：

  • spring boot 2.6.0
  • redisson-spring-boot-starter 3.28.0
一、加入依赖&配置
 <dependency>
        <groupId>org.redisson</groupId>
        <artifactId>redisson-spring-boot-starter</artifactId>
        <version>3.28.0</version>
</dependency>

  • 1
  • 2
  • 3
  • 4
  • 5

application.properties

spring.application.name=springboot-redis-delayed-queue-demo

spring.redis.database=2
spring.redis.host=localhost
spring.redis.password=123456
spring.redis.port=6379

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
二、延时任务添加
package cn.aohan.delayedqueue.provider;

import cn.aohan.delayedqueue.model.DelayedTaskInfo;
import cn.aohan.delayedqueue.model.TaskData;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.redisson.codec.JsonJacksonCodec;
import org.springframework.stereotype.Component;

import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.function.Predicate;

/**
 * @author 傲寒
 * @date 2024/4/19
 */
@Component
public class DelayedQueueProvider {

    private final RedissonClient redissonClient;

    public DelayedQueueProvider(RedissonClient redissonClient) {
        this.redissonClient = redissonClient;
    }

    /**
     * 添加延迟任务
     *
     * @param delayedName 延迟名称
     * @param val         值
     * @param delayTime   延迟时间
     * @param timeUnit    时间单位
     */
    public void addDelayedTask(String delayedName, TaskData val, long delayTime, TimeUnit timeUnit) {
        final DelayedTaskInfo task = new DelayedTaskInfo();
        task.setCreateAt(System.currentTimeMillis());
        task.setDelayTime(delayTime);
        task.setTimeUnit(timeUnit);
        task.setVal(val);
        task.setDelayedName(delayedName);
        final RDelayedQueue<DelayedTaskInfo> delayedQueue = getDelayedQueue(delayedName);
        delayedQueue.offer(task, delayTime, timeUnit);
    }

    /**
     * 删除任务
     *
     * @param queueName 队列名称
     * @param taskId    任务id
     */
    public void removeTask(String queueName, String taskId) {
        final RBlockingDeque<DelayedTaskInfo> blockingDeque = getBlockingDeque(queueName);
        final Predicate<DelayedTaskInfo> predicate = item -> {
            final TaskData val = item.getVal();
            return Objects.nonNull(val) && Objects.equals(taskId, val.taskId);
        };
        blockingDeque.removeIf(predicate);
        final RDelayedQueue<DelayedTaskInfo> delayedQueue = getDelayedQueue(getBlockingDeque(queueName));
        delayedQueue.removeIf(predicate);
    }


    /**
     * 获取阻塞deque
     *
     * @param queueName 队列名称
     * @return {@link RBlockingDeque}<{@link DelayedTaskInfo}>
     */
    public RBlockingDeque<DelayedTaskInfo> getBlockingDeque(String queueName) {
        return redissonClient.getBlockingDeque(queueName, JsonJacksonCodec.INSTANCE);
    }

    /**
     * 获取延迟队列
     *
     * @param queueName 队列名称
     * @return {@link RDelayedQueue}<{@link DelayedTaskInfo}>
     */
    private RDelayedQueue<DelayedTaskInfo> getDelayedQueue(String queueName) {
        return redissonClient.getDelayedQueue(getBlockingDeque(queueName));
    }

    /**
     * 获取延迟队列
     *
     * @param blockingDeque 阻塞deque
     * @return {@link RDelayedQueue}<{@link DelayedTaskInfo}>
     */
    private RDelayedQueue<DelayedTaskInfo> getDelayedQueue(RBlockingDeque<DelayedTaskInfo> blockingDeque) {
        return redissonClient.getDelayedQueue(blockingDeque);
    }

}


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
三、监听延时任务到期

延时队列名称常量

/**
 * @author 傲寒
 * @date 2024/4/19
 */
public class QueueConstant {

    /**
     * 测试延迟任务队列 name
     */
    public static final String TEST_DELAYED_TASK_QUEUE = "test_delayed_task_queue";


}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13

listener监听

package cn.aohan.delayedqueue.listener;

import cn.aohan.delayedqueue.constant.QueueConstant;
import cn.aohan.delayedqueue.model.DelayedTaskInfo;
import cn.aohan.delayedqueue.provider.DelayedQueueProvider;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * 延迟任务监听器
 *
 * @author 傲寒
 * @date 2024/4/19
 */
@RequiredArgsConstructor
@Slf4j
@Component
public class DelayedTaskListener implements ApplicationRunner {

    private final DelayedQueueProvider delayedQueueProvider;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        delayedTaskHandle(QueueConstant.TEST_DELAYED_TASK_QUEUE);
    }

    public void delayedTaskHandle(String delayedQueueName) {
        final Thread thread = new Thread(() -> {
            final RBlockingDeque<DelayedTaskInfo> blockingDeque = delayedQueueProvider.getBlockingDeque(delayedQueueName);
            while (true) {
                try {
                    //将到期的数据取出来，等待超时
                    final DelayedTaskInfo delayedTaskInfo = blockingDeque.poll(2, TimeUnit.MINUTES);
                    if (Objects.isNull(delayedTaskInfo)) {
                        continue;
                    }
                    log.info("DelayedTask task :[{}]", delayedTaskInfo);
                } catch (Exception e) {
                    log.error("DelayedTaskListener#delayedTaskHandle error delayedQueueName:[{}]", delayedQueueName, e);
                }
            }
        });
        thread.setDaemon(true);
        thread.start();
    }
}


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
四、测试延时任务添加
package cn.aohan.delayedqueue.controller;

import cn.aohan.common.dto.Result;
import cn.aohan.delayedqueue.constant.QueueConstant;
import cn.aohan.delayedqueue.model.dto.TestDelayedDTO;
import cn.aohan.delayedqueue.provider.DelayedQueueProvider;
import lombok.AllArgsConstructor;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author 傲寒
 * @date 2024/4/19
 */
@AllArgsConstructor
@RestController
@RequestMapping("/api/test/delayed")
public class DelayedQueueTestController {

    private final DelayedQueueProvider delayedQueueProvider;

    /**
     * 添加延迟任务
     *
     * @param delayedTask 延迟任务
     * @return {@link Result}<{@link Void}>
     */
    @PostMapping
    public Result<Void> addDelayedTask(@RequestBody TestDelayedDTO delayedTask) {
        delayedQueueProvider.addDelayedTask(
                QueueConstant.TEST_DELAYED_TASK_QUEUE,
                delayedTask.getVal(),
                delayedTask.getDelayTime(),
                delayedTask.getTimeUnit()
        );
        return Result.success();
    }

}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
五、大致流程及其原理

在一开始创建延时队列的时候会创建一个QueueTransferTask

org.redisson.RedissonDelayedQueue#RedissonDelayedQueue

请添加图片描述
请添加图片描述

channelName = prefixName("redisson_delay_queue_channel", getRawName());

  • 这里并订阅（channel）延时队列创建任务调度(主要是使用netty中时间轮)。
  • 使用pushTaskAsync去操作lua脚本移除redis 中LIST和ZSET的元素。

根据延迟时间插入到对中合适的位置，主要是

org.redisson.RedissonDelayedQueue#offerAsync 方法中的一段lua脚本

local value = struct.pack('Bc0Lc0', string.len(ARGV[2]), ARGV[2], string.len(ARGV[3]), ARGV[3]);
redis.call('zadd', KEYS[2], ARGV[1], value);
redis.call('rpush', KEYS[3], value);
local v = redis.call('zrange', KEYS[2], 0, 0); 
if v[1] == value then 
   redis.call('publish', KEYS[4], ARGV[1]); 
end;

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7

总而言之，这段代码的功能是:

  • 将两个字符串打包成一个二进制值。
  • 将打包后的值添加到一个排序集合(zset)中，并为其指定分数(当前时间+延迟时间)。
  • 将同样的值添加到一个列表的尾部。
  • 如果添加的元素是排序集合中的第一个元素，则向发布一条消息(上边的订阅的channel)。

然后使用BLPOP阻塞的去获取LIST的元素

redisson实现延迟队列的原理，简单来说，将数据插入到延迟队列时，会存入到延迟队列的list和zset结构中，通过任务调度的方式将延迟队列中到期的数据取出，然后放入到阻塞队列中，客户端通过BLPOP的命令阻塞的拉取阻塞队列的数据，若拉取到数据就可以进行业务逻辑的处理。

项目源码

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/秋刀鱼在做梦/article/detail/823197
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号