（四）延时任务篇——redisson实现延迟任务实战

news2024/11/15 13:52:48

前言

上一节内容中介绍了如何使用redis的zset结构实现延迟任务的实战内容，从使用角度来说还是略显繁琐，而且定时任务的方式扫描redis获取过去的任务也会存在任务空转的问题。在此基础上，我们可以使用redisson的阻塞队列，完成延迟任务的实战场景。

正文

引入redisson的pom依赖

<dependency>
	<groupId>org.redisson</groupId>
	<artifactId>redisson-spring-boot-starter</artifactId>
</dependency>

配置redisson的阻塞队列和延迟队列

package com.yundi.tps.config;

import org.redisson.Redisson;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

@Component
public class RedissionConfig {
    /**
     * 对列名称
     */
    private final String queueName = "orderQueue";

    @Bean(destroyMethod = "shutdown")
    public RedissonClient redissonClient() {
        Config config = new Config();
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");
        return Redisson.create(config);
    }


    //阻塞队列
    @Bean
    public RBlockingQueue<String> blockingQueue(RedissonClient redissonClient) {
        return redissonClient.getBlockingQueue(queueName);
    }


    //延迟队列
    @Bean
    public RDelayedQueue<String> delayedQueue(RedissonClient redissonClient, RBlockingQueue<String> blockQueue) {
        return redissonClient.getDelayedQueue(blockQueue);
    }
}

创建一个订单消费任务，用于监控阻塞队列中的消息，消费延时任务消息

package com.yundi.tps.config;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.springframework.stereotype.Component;


/**
 * @date 2023/8/30 15:09
 */
@Slf4j
@Component
public class OrderTask {
    @Resource
    private RBlockingQueue<Object> blockingQueue;

    @PostConstruct
    public void take() {
        new Thread(() -> {
            while (true) {
                log.info("开始消费");
                try {
                    //将到期的数据取出来，如果一直没有到期数据，就一直等待。
                    log.info(blockingQueue.take().toString());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

}

模拟任务发送，创建一个订单的请求

    @Operation(summary = "创建订单-redisson")
    @PostMapping("saveOrderByRedisson")
    public ApiResponse saveOrderByRedisson() {
        String orderId = String.valueOf(IdWorker.getId());
        delayedQueue.offerAsync(orderId, 5, TimeUnit.SECONDS);
        return ApiResponse.ok();
    }