基于redission的redis延迟队列实现_redission延时队列

在工作中，我们经常会遇到一些场景，比如订单到期未支付导致取消，或者到期后自动续费等。在这些情况下，我们发现延迟队列非常适合使用。常见的延迟队列实现包括rabbitMQ的死信队列和RocketMQ的延迟队列。然而，有时候项目规模较小，没有引入消息中间件，但又需要使用延迟队列的场景。在这种情况下，我们可以利用已有的redis实现的延迟队列来解决问题。
所以今天介绍Redisson的延迟队列，Redisson有一个RDelayedQueue是一个专门用于处理延迟任务的队列，RDelayedQueue不会阻止生产者向已满的队列中添加元素，而是会将它们放入一个等待队列中，直到有空间可用。允许元素在一定时间后被重新入队，或者在指定的时间间隔内被消费。下面我们来实现延迟队列：
核心思想是通过线程执行redission的延时队列（DelayedQueue），通过take()方法尝试从队列中取出元素。

引入redission jar包：

 <dependency>
            <groupId>org.redisson</groupId>
            <artifactId>redisson</artifactId>
            <version>3.17.4</version>
        </dependency>
1
2
3
4
5

配置

先配置项目中的redission 配置文件：

@Configuration
public class RedissonConfig {

    @Value("${spring.redis.host}")
    private String redisHost;
    @Value("${spring.redis.port}")
    private String redisPort;
    @Value("${spring.redis.password}")
    private String password;

    @Bean(destroyMethod = "shutdown")
    public RedissonClient redissonClient() {
        Config config = new Config();
        config.useSingleServer().setAddress("redis://" + redisHost + ":" + redisPort);
        if(StringUtils.isBlank(password)) {
            config.useSingleServer().setPassword(null);
        } else {
            config.useSingleServer().setPassword(password);
        }
        return Redisson.create(config);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

也可以使用redission springboot 集成配置，引入redisson-spring-data-2x，这样就无需配置类了。详见官网：

https://github.com/redisson/redisson/tree/master/redisson-spring-boot-starter

基于redission api的服务

增加封装操作api的方法，提供crud队列元素功能：

@Service
public class RedissonDelayedQueueService {

    @Autowired
    private RedissonClient redissonClient;

    public <E> void addQueue(E e, long delay, TimeUnit timeUnit, String queueName) {
        RBlockingDeque<E> blockingDeque = redissonClient.getBlockingDeque(queueName);
        RDelayedQueue<E> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
        delayedQueue.offer(e, delay, timeUnit);
    }

    public <E> RBlockingDeque<E> getQueue(String queueName) {
        RBlockingDeque<E> blockingDeque = redissonClient.getBlockingDeque(queueName);
        redissonClient.getDelayedQueue(blockingDeque);
        return blockingDeque;
    }

    public <E> void removeQueueElement(E e, String queueName) {
        RBlockingDeque<E> blockingDeque = redissonClient.getBlockingDeque(queueName);
        RDelayedQueue<E> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
        delayedQueue.remove(e);
    }

}
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

• add为添加一个元素到队列方法；
• get 为获取一个队列
• removeQueueElement 为删除队列中的元素，适用于某些想直接删除已经添加的键值场景，比如用户下单没有立即完成付款，系统规定30分钟内完成支付，否则按废单处理，这时用户在5分钟时完成了付款，则需要直接删除当前用户的延时任务。

订单的阻塞队列

现在我们就可以实现我们一开始提到的核心思想，比如我们想实现一个订单的阻塞队列监听：
先定义一个键值：

public interface CacheKeyDefinition {
 /**
     * 订单延时到生效时间执行
     */
    String REDIS_KEY_ORDER = "redis_key_publish_order";
}
1
2
3
4
5
6

@Slf4j
@Component
public class OrderDelayedQueueListener {
    @Resource
    private RedissonDelayedQueueService redissonDelayedQueueService;
    @Resource
    private OrderHandler orderHandler;
    private final ExecutorService singlePoolExecutor
            = Executors.newSingleThreadExecutor();
    /**
     * 每次
     */
    @PostConstruct
    public void listener() {
        singlePoolExecutor.execute(()->{
            while (true){
                RBlockingDeque<String> blockingDeque = redissonDelayedQueueService.getQueue(CacheKeyDefinition.REDIS_KEY_ORDER);
                String priceListCode  = null;
                try {
                    id = blockingDeque.take();
                } catch (InterruptedException e) {
                    log.error("消费异常",e);
                }
                log.info("获取到订单队列:{}",id);
                orderHandler.handle(id);
            }
        });
    }
}
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

在实际业务中使用示例：

public class OrderserviceImpl{
	@Autowired
	private  RedissonDelayedQueueService delayedQueueService;
	
	void create(){
	//创建订单
	String id=;
	//将当前订单放入延时任务中：
	 delayedQueueService.addQueue(id, 30, TimeUnit.MINUTES, CacheKeyDefinition.REDIS_KEY_ORDER);
	
}
	
	//付款
	void pay(String id){
		//判定付款开始。。
		//判定付款成功。。
		
		 //删除redis 延迟队列元素
        delayedQueueService.removeQueueElement(id, CacheKeyDefinition.REDIS_KEY_ORDER);
	}
}
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

订单的延时处理

class OrderHandler{
	void handle(String id){
		//关闭当前订单
		close(id);
	}
}
1
2
3
4
5
6