即使不是做电商业务的同学，也一定知道订单超时关闭这种业务场景，这个场景大致就是用户下单后，如果在一定时间内未支付（比如15分钟、半小时），那么系统就会把这笔订单给关闭掉。这个功能实现的方式有很多种，比如JDK中自带的DelayQueue延迟队列、Timer、ScheduledThreadPoolExecutor，强烈推荐的RocketMQ、RabblitMQ及Kafka等消息队列，还有就是Hutool的SystemTimer、Netty的HashedWheelTimer等等，感兴趣的可以去了解一下。今天我们就先看看Redis和Redisson是如何实现延迟消息的。
一、Redis如何实现延迟消息
1.1 过期key监听
很多人都知道Redis有一个过期监听的功能，在redis.conf中加一条notify-keyspace-events开启过期监听，然后在代码中实现KeyExpirationMessageListener就可以监听key的过期消息了。
这样就可以在接收到过期消息的时候进行关单操作了，但是这个方案并不推荐，Redis官方明确说过Redis并不保证key在过期的时候就能被立即删除，更不保证这个消息能被立即发出，所以，消息延迟是必然的，数据量越大延迟的时间越长。
而且，在Redis5.0之前，这个消息是通过PUB/SUB模式发出的，不会做持久化，至于你有没有接收到，有没有消费成功，它不管，所以，如果发消息的时候客户端挂了，之后再恢复的话，这个消息也就彻底丢了。
1.2 Zset
我们可以借助Redis中的有序集合——zset来实现这个功能，zset是一个有序集合，每一个元素（member）都关联了一个score，可以通过score排序来取集合中的值。
这么实现的优点就是可以借助redis持久化和高可用机制，避免数据丢失。
1.3 zset实现超时关单Demo
步骤：
导入依赖：
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>Demo：
@Service
public class OrderDelayService {
@Autowired
private RedisTemplate<String,String> redisTemplate;
//延迟15分钟
private final long delayTime = 15*60*1000;
//订单号key前缀
private static final String ORDER_KEY="order:";
//延迟关单key
private static final String DELAY_KEY="close_orders";
/**
* 创建订单
* @param orderId
*/
public void createOrder(String orderId){
//...
//创建订单成功
//1.获取当前时间戳
long currentTime = System.currentTimeMillis();
//2.score:当前时间+延迟时间
long score=currentTime+delayTime;
//3.加入redis zset集合
redisTemplate.opsForZSet().add(
DELAY_KEY, //redis key
ORDER_KEY+orderId, //member
score //score
);
}
/**
* 任务间隔一秒执行一次
*/
@Scheduled(fixedDelay = 1000)
public void closeExpiredOrders(){
//当前时间戳
long currentTime = System.currentTimeMillis();
ZSetOperations<String, String> zSetOps = redisTemplate.opsForZSet();
//取出所有数据(已排好序）
Set<String> orderKeys = zSetOps.range(DELAY_KEY, 0, -1);
for (String orderKey : orderKeys) {
//获取score，
double score = zSetOps.score(DELAY_KEY, orderKey);
if(currentTime>=score){
String orderId = orderKey.substring(ORDER_KEY.length());
//进行关单操作...closeOrder(orderId)
//从zset里移除该订单
zSetOps.remove(DELAY_KEY,orderKey);
}
}
}
}二、Redission如何实现延迟消息
2.1 实现原理
Redission中定义了分布式延迟队列DelayedQueue，其实就是在zset的基础上增加了一个基于内存的延迟队列。
大致的流程如下：
2.2 案例
导入依赖：
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson-spring-boot-starter</artifactId>
<version>3.13.1</version>
</dependency>定义一个客户端：
@Component
public class RedissionConfig {
@Bean(destroyMethod = "shutdown")
public RedissonClient redssion(){
Config config=new Config();
config.useSingleServer().setAddress("redis://127.0.0.1");
return Redisson.create(config);
}
}实现：
@Component
public class RedissionDelay {
@Autowired
private RedissonClient client;
/**
* 创建订单，并设置过期时间
* @param orderId
*/
public void createOrder(String orderId){
//...
//创建订单成功
RBlockingDeque<Object> blockingDeque = client.getBlockingDeque("orderQueue");
RDelayedQueue<Object> delayedQueue = client.getDelayedQueue(blockingDeque);
//将订单加入延迟队列，延迟时间为15分钟
delayedQueue.offer(orderId,15, TimeUnit.MINUTES);
}
/**
* 关单操作
*/
public void closeOrder(){
RBlockingDeque<Object> orderQueue = client.getBlockingDeque("orderQueue");
new Thread(() -> {
while (true){
try {
String orderId = (String) orderQueue.take();
//进行关单操作，closeOrder(orderId)
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
}上述例子，我们用RDelayedQueue的offer方法将订单添加到了延迟队列，并指定了延迟时间，当元素的延迟时间到达时，Redission会将元素从RDelayedQueue转移到与之关联的RBlockingDeque。
然后在检查是否要关单的时候，另起了一个线程，不断循环读取到期的订单。值得注意的是 take方法从RBlockingDeque中获取元素，这是一个阻塞操作，如果没有元素，它会一直等到，直到有元素。
End：希望对大家有所帮助，如果有纰漏或者更好的想法，请您一定不要吝啬你的赐教🙋。
