【教程】Redis实现延迟队列 Redisson_redisson延迟队列数据在redis中看不到

【教程】Redis实现延迟队列 Redisson

前提：需要提前集成Redis

1 添加依赖

        <dependency>
            <groupId>org.redisson</groupId>
            <artifactId>redisson</artifactId>
            <version>3.19.3</version>
        </dependency>

2. 创建Redisson 配置类

 
 
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.redisson.config.SingleServerConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
/**
 * @description Redisson 配置类
 */
@Configuration
public class MyRedissonConfig {
 
    @Value(value = "${spring.redis.host}")
    private String host;
    @Value(value = "${spring.redis.port}")
    private int port;
    @Value(value = "${spring.redis.database}")
    private int database;
    @Value(value = "${spring.redis.password}")
    private String password;
 
    /**
     * 单Redis节点模式配置方法
     * 其他配置參數，看:
     * <a href = "https://github.com/redisson/redisson/wiki/2.-%E9%85%8D%E7%BD%AE%E6%96%B9%E6%B3%95#26-%E5%8D%95redis%E8%8A%82%E7%82%B9%E6%A8%A1%E5%BC%8F">
     * 单Redis节点模式配置方法
     * </a>
     *
     * @return {@link RedissonClient}
     */
    @Bean(destroyMethod = "shutdown")
    RedissonClient redisson() {
        Config config = new Config();
        //Redis多节点
        // config.useClusterServers()
        //     .addNodeAddress("redis://127.0.0.1:6379", "redis://127.0.0.1:7001");
        //Redis单节点
        SingleServerConfig singleServerConfig = config.useSingleServer();
        //可以用"rediss://"来启用SSL连接
        String address = "redis://" + host + ":" + port;
        singleServerConfig.setAddress(address);
        //设置 数据库编号
        singleServerConfig.setDatabase(database);
        singleServerConfig.setPassword(password);
        //连接池大小:默认值：64
        // singleServerConfig.setConnectionPoolSize()
        return Redisson.create(config);
    }
 
}
 

3.创建 RedissonDelayQueue

 
import lombok.extern.slf4j.Slf4j;
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.redisson.config.SingleServerConfig;
import org.springframework.stereotype.Component;
 
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;
 
@Component
@Slf4j
public class RedissonDelayQueue {
 
    @Resource
    private RedissonClient redissonClient;
 
    private RDelayedQueue<String> delayQueue;
    private RBlockingQueue<String> blockingQueue;
 
    @PostConstruct
    public void init() {
        initDelayQueue();
        startDelayQueueConsumer();
    }
 
    /**
     * 初始化
     */
    private void initDelayQueue() {
//        Config config = new Config();
//        SingleServerConfig serverConfig = config.useSingleServer();
//        serverConfig.setAddress("redis://localhost:6379");
//        redissonClient = Redisson.create(config);
 
        blockingQueue = redissonClient.getBlockingQueue("SANYOU");
        delayQueue = redissonClient.getDelayedQueue(blockingQueue);
    }
 
    /**
     * 监听延时任务
     */
    private void startDelayQueueConsumer() {
        new Thread(() -> {
            while (true) {
                try {
                    String task = blockingQueue.take();
                    log.info("接收到延迟任务:{}", task);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }, "SANYOU-Consumer").start();
    }
 
    /**
     * 添加延迟任务
     * @param task
     * @param seconds
     */
    public void offerTask(String task, long seconds) {
        log.info("添加延迟任务:{} 延迟时间:{}s", task, seconds);
        delayQueue.offer(task, seconds, TimeUnit.SECONDS);
    }
    
    
    
 
}

4. 使用测试

    /**
     * redis延迟队列
     */
    @Resource
    private RedissonDelayQueue redissonDelayQueue;
 
    /**
     * 添加延迟队列
     * @param task
     */
    @GetMapping("/add")
    public void addTask(@RequestParam("task") String task, int time) {
        redissonDelayQueue.offerTask(task, time);
    }

请求

http://127.0.0.1:8080/add?task=你好&time=7

结果