Redis过期回调事件实现延迟队列：配置与代码编写详解
4.Redis过期回调
Redis的key过期回调事件也可以达到延迟排队的效果。简单来说，我们开启监听key是否过期的事件。一旦密钥过期，就会触发回调事件。
修改redis.conf文件以启用notify-keyspace-events Ex
notify-keyspace-events Ex
Redis监听配置，注入Bean RedisMessageListenerContainer
@Configuration
public class RedisListenerConfig {
@Bean
RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
return container;
}
}
编写Redis过期回调监听方法，必须继承KeyExpirationEventMessageListener，这有点类似于MQ消息监听。
@Component
public class RedisKeyExpirationListener extends KeyExpirationEventMessageListener {
public RedisKeyExpirationListener(RedisMessageListenerContainer listenerContainer) {
super(listenerContainer);
}
@Override
public void onMessage(Message message, byte[] pattern) {
String expiredKey = message.toString();
System.out.println("监听到key：" + expiredKey + "已过期");
}
}
代码就写到这里了。这很简单。接下来测试一下效果。在redis-cli客户端添加key，设置过期时间为3秒。
set xiaofu 123 ex 3
已在控制台成功监控到此过期密钥。
检测到的过期密钥是：xiaofu
5.RabbitMQ延迟队列
使用RabbitMQ作为延迟队列是一种比较常见的方式。其实RabbitMQ本身并不直接支持延迟队列功能，而是通过RabbitMQ消息队列的TTL和DXL这两个属性间接实现的。
我们先来了解一下TTL和DXL这两个概念：
生存时间（TTL）：
TTL，顾名思义：指的是消息的生存时间。 RabbitMQ可以通过x-message-tt参数设置消息在指定Queue（队列）和Message（消息）上的生存时间。它的值是以微米为单位的非负整数。第二。
RabbitMQ可以从队列和消息本身两个维度来设置消息过期时间。
如果队列的TTL和队列中的消息的TTL同时设置，则TTL值取两者中较小的值。队列中的消息在队列中存在的时间超过了TTL过期时间，就成为死信。
死信交换（DLX）
DLX是一个死信开关，它是一个死信队列绑定到一个死信开关。 RabbitMQ的Queue可以配置两个参数，x-dead-letter-exchange和x-dead-letter-routing-key（可选）。一旦队列中出现死信，就会使用这两个参数。参数可以将消息重新路由到另一个Exchange（交换机），以便可以再次消费该消息。
x-dead-letter-exchange：队列中出现死信后，该死信将被重新路由并转发到指定的交换机（交换机）。
x-dead-letter-routing-key：指定要发送的routing-key，通常是要转发的队列。
队列中出现Dead Letter的情况包括：
下图看如何实现超过30分钟未支付的平仓功能。我们将订单消息A0001发送到延迟队列order.delay.queue，并将x-message-tt消息生存时间设置为30分钟。当到达 30 分钟时，几分钟后，订单消息 A0001 变为 Dead。 Letter（死信），延迟队列检测到死信。通过配置x-dead-letter-exchange，将死信重新转发到可以正常消费的订单队列，并且可以直接监听订单队列来处理订单逻辑。
指定发送消息时消息将延迟多长时间
public void send(String delayTimes) {
amqpTemplate.convertAndSend("order.pay.exchange", "order.pay.queue","大家好我是延迟数据", message -> {
// 设置延迟毫秒值
message.getMessageProperties().setExpiration(String.valueOf(delayTimes));
return message;
});
}
}
设置延迟队列中出现死信后的转发规则
/**
* 延时队列
*/
@Bean(name = "order.delay.queue")
public Queue getMessageQueue() {
return QueueBuilder
.durable(RabbitConstant.DEAD_LETTER_QUEUE)
// 配置到期后转发的交换
.withArgument("x-dead-letter-exchange", "order.close.exchange")
// 配置到期后转发的路由键
.withArgument("x-dead-letter-routing-key", "order.close.queue")
.build();
}
6.时间轮
前面的延迟队列的实现方法都比较简单易懂，而时间轮算法就有点抽象了。 Kafka和Netty都是基于时间轮算法实现延迟队列。下面主要练习Netty的延迟队列，讲解时间轮的原理。
我们先看一个时间轮的示意图，解释一下时间轮的一些基本概念。
轮：时间轮。图中的圆盘可以看作是钟表的刻度。例如，一轮的长度是24秒，刻度数是8，那么每个刻度代表3秒。那么时间精度是3秒。时间长度/刻度值越大，精度越高。
添加定时或延迟任务A时，如果延迟25秒后执行，但时间轮轮次长度只有24秒，那么根据时间轮的长度和规模。指针位置索引意味着任务A将绕一圈并指向0格。此时时间轮会记录任务的轮次和索引信息。当round=0，index=0时，指针指向0号格，任务A不会被执行，因为round=0不满足要求。
所以每个格子代表一段时间，比如1秒、25秒就会指向0号格，任务就放在每个格子对应的链表中，这有点类似于HashMap的数据。
Netty主要使用HashedWheelTimer来构建延迟队列。 HashedWheelTimer的底层数据结构仍然使用DelayedQueue，但它是使用时间轮算法实现的。
接下来我们使用Netty来简单实现一个延迟队列。有许多 HashedWheelTimer 构造函数。我们来解释一下每个参数的含义。
ThreadFactory：表示用于生成工作线程，一般使用线程池；
tickDuration和unit：每个网格的时间间隔，默认100ms；
ticksPerWheel：一圈的刻度数，默认为512。如果传入的值不是2的N次方，则会调整为大于等于参数2的N次方的值，即有利于优化哈希值的计算。
public HashedWheelTimer(ThreadFactory threadFactory, long tickDuration, TimeUnit unit, int ticksPerWheel) {
this(threadFactory, tickDuration, unit, ticksPerWheel, true);
}
TimerTask：定时任务的实现接口，其中run方法包装了定时任务的逻辑。
Timeout：定时任务提交给Timer后返回的句柄。通过这个句柄，可以从外部取消定时任务，并对定时任务的状态进行一些基本的判断。
Timer：是HashedWheelTimer实现的父接口。它只是定义了如何提交定时任务以及如何停止整个定时机制。
public class NettyDelayQueue {
public static void main(String[] args) {
final Timer timer = new HashedWheelTimer(Executors.defaultThreadFactory(), 5, TimeUnit.SECONDS, 2);
//定时任务
TimerTask task1 = new TimerTask() {
public void run(Timeout timeout) throws Exception {
System.out.println("order1 5s 后执行 ");
timer.newTimeout(this, 5, TimeUnit.SECONDS);//结束时候再次注册
}
};
timer.newTimeout(task1, 5, TimeUnit.SECONDS);
TimerTask task2 = new TimerTask() {
public void run(Timeout timeout) throws Exception {
System.out.println("order2 10s 后执行");
timer.newTimeout(this, 10, TimeUnit.SECONDS);//结束时候再注册
}
};
timer.newTimeout(task2, 10, TimeUnit.SECONDS);
//延迟任务
timer.newTimeout(new TimerTask() {
public void run(Timeout timeout) throws Exception {
System.out.println("order3 15s 后执行一次");
}
}, 15, TimeUnit.SECONDS);
}
}
从执行结果来看，order3和order3延迟任务只执行一次，而order2和order1是定时任务，按照不同的周期重复执行。
order1将在5秒后执行
order2将在10秒后执行
order3 15s后执行一次
order1将在5秒后执行
order2将在10秒后执行
总结
为了让大家更容易理解，上面的代码比较简单粗暴。几种实现方法的demo已经提交到github地址：感兴趣的朋友可以下载运行。
我为这篇文章工作了很长时间。写作并不比工作轻松。我查阅了资料并反复验证了demo的可行性，搭建了各种RabbitMQ和Redis环境。我只想说太难了！
这篇关于Java的6种延迟队列的实现方法的文章就一口气讲完了（连面试官都得服气）。关于Java延迟队列的更多信息，请搜索Script House之前的文章或者继续浏览下面的相关文章。希望大家以后多多支持Script House！
phper