java

关注公众号 jb51net

关闭
首页 > 软件编程 > java > springboot redisson延时队列

springboot整合redisson实现延时队列(附仓库地址)

作者：ゞ註﹎錠oo

延时队列用于管理需要定时执行的任务,对于大数据量和高实时性需求,使用延时队列比定时扫库更高效,Redisson提供一种高效的延时队列实现方式,本文就来详细的介绍一下,感兴趣都可以了解学习

应用场景

通常在一些需要经历一段时间或者到达某个指定时间节点才会执行的功能，比如以下这些场景：

  • 订单超时提醒
  • 收货自动确认
  • 会议提醒
  • 代办事项提醒

为什么使用延时队列

对于数据量小且实时性要求不高的需求来说，最简单的方法就是定时扫描数据库。

但是，当数量达到数百万、上千万级别且时，定时扫库就显得非常低效且消耗资源，

甚至有些时间间隔小实时性要求高的情况，上一次扫描还没结束，下一次就又开始了，

这时候如果使用延时队列就会比较合适

延时队列的几种方式：

Redisson 实现延时队列

顾名思义 Redis son 就是 Redis 的儿子，举个栗子先：

1.引入 pom

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>${lastest.version}</version>
</dependency>

2.封装一个 RedissonQueue 类

@Service
public class RedissonQueue {

    public static final String QUEUE = "delayQueue";

    // 默认超时时间，30秒
    public static final Integer DEFAULT_TIMEOUT = 30;

    @Resource
    private RedissonClient redissonClient;

    // 加入任务并设置到期时间
    public void offer(String taskId, Integer timeout) {
        RDelayedQueue<String> delayedQueue = delayedQueue();
        delayedQueue.offer(taskId, Objects.isNull(timeout) ? DEFAULT_TIMEOUT : timeout, TimeUnit.SECONDS);
    }

    // 移除任务
    public void remove(String taskId) {
        RDelayedQueue<String> delayedQueue = delayedQueue();
        delayedQueue.removeIf(messageId -> messageId.equals(taskId));
    }

    // 任务列表
    public RDelayedQueue<String> delayedQueue() {
        RBlockingDeque<String> blockingDeque = blockingDeque();
        return redissonClient.getDelayedQueue(blockingDeque);
    }

    public RBlockingDeque<String> blockingDeque() {
        return redissonClient.getBlockingDeque(QUEUE);
    }

    public boolean isShutdown() {
        return redissonClient.isShutdown();
    }

    public void shutdown() {
        redissonClient.shutdown();
    }

}

3.交给 Spring 管理

@Slf4j
@Service
public class RedissonService implements ApplicationRunner {

    @Resource
    private RedissonQueue redissonQueue;

    @Resource(name = "threadPoolTaskExecutor")
    private ThreadPoolTaskExecutor executor;

    @Override
    public void run(ApplicationArguments args) {
        RBlockingDeque<String> blockingDeque = redissonQueue.blockingDeque();
        executor.execute(() -> {
            while (true) {
                if (redissonQueue.isShutdown()) {
                    return;
                } else {
                    String messageId = null;
                    try {
                        messageId = blockingDeque.take();
                    } catch (InterruptedException e) {
                        log.warn("RedissonConsumer error:{}", e.getMessage());
                    }
                    if (!Objects.isNull(messageId) && !messageId.isEmpty()) {
                        log.warn("timeout messageId : {}", messageId);
                    }
                }
            }
        });

    }

    // 初始化，启动服务就执行一次
    @PostConstruct
    public void init() {
        redissonQueue.delayedQueue();
    }

    @PreDestroy
    public void shutdown() {
        redissonQueue.shutdown();
    }

}

4.测试接口

@Operation(summary = "添加任务", description = "添加任务")
@PostMapping
public ResponseEntity<?> add(@RequestParam(value = "taskId", required = false) String taskId,
                             @RequestParam(value = "timeout", required = false) Integer timeout) {
    taskId = StringUtils.isEmpty(taskId) ? String.valueOf(snowflake.nextId()) : taskId;
    redissonQueue.offer(taskId, timeout);
    return ResponseEntity.ok().body(redissonQueue.delayedQueue());
}

@Operation(summary = "移除任务", description = "移除任务")
@DeleteMapping(value = "/{taskId}")
public ResponseEntity<?> remove(@PathVariable("taskId") String taskId) {
    redissonQueue.remove(taskId);
    return ResponseEntity.ok().body(redissonQueue.delayedQueue());
}

5.测试结果

添加10个任务

在这里插入图片描述

删除第1个任务

在这里插入图片描述

可以看到第一个任务删除后没有被执行（没有设置到期时间，默认为30秒到期）

在这里插入图片描述

实现原理

lua 脚本主要干两件事

当客户端监听到 redisson_delay_queue_channel:delayQueue 这个 channel 的消息时，会再次提交一个客户端延迟任务，延迟时间就是消息（最早到期时间任务的到期时间戳）当前时间戳
这个时间其实也就是 redisson_delay_queue_channel:delayQueue 中最早到期时间的任务的剩余的延迟时间。
一旦时间来到最早到期时间任务的到期时间戳，redisson_delay_queue_timeout:delayQueue 中最早到期时间的任务已经到期，客户端的延迟任务也同时到期，
于是开始执行 lua 脚本操作，及时将到期任务放到目标队列中。然后再次发布剩余的延迟任务中最早到期任务的到期时间戳到 channel
中，
如此循环运行下去，保证 redisson_delay_queue_timeout:delayQueue 中到期数据能及时放到目标队列中。
这里存在一个特殊情况，需要项目启动时就执行一次延时队列。因为由于没有客户端延迟任务的执行，
可能会出现 redisson_delay_queue_timeout:delayQueue 队列中有到期但是没有被放到目标队列的可能，启动就执行一次是为了保证到期的数据能被及时放到目标队列中。

结论

Redisson 这种实现方案是比较合适且靠谱的，一般中小型项目建议用 Redisson 实现延迟队列，规模较大的项目直接上 MQ。

整合DEMO仓库地址

到此这篇关于springboot整合redisson实现延时队列(附仓库地址)的文章就介绍到这了,更多相关springboot redisson延时队列内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:
阅读全文