文章目录
- 延时队列
- JDK自带的延时队列实现
- Redis实现延迟队列
- RabbitMQ 延时队列
延时队列
延时队列是一种特殊类型的队列,它允许元素在特定时间间隔后才能被处理。这种队列在处理具有延迟需求的任务时非常有用,例如定时任务、事件驱动系统等
延时队列在项目中的应用还是比较多的,尤其像电商类平台:
1、订单成功后,在30分钟内没有支付,自动取消订单
2、外卖平台发送订餐通知,下单成功后60s给用户推送短信。
3、如果订单一直处于某一个未完结状态时,及时处理关单,并退还库存
4、淘宝新建商户一个月内还没上传商品信息,将冻结商铺等
JDK自带的延时队列实现
在Java中,DelayQueue 是一个无界阻塞队列,它存放实现了 Delayed 接口的对象,只有在延迟时间到了,对象才能从队列中取出来。DelayQueue 是基于优先队列实现的,队列中的元素按照延迟时间的先后顺序排序。
DelayQueue是一个BlockingQueue(无界阻塞)队列,它本质就是封装了一个PriorityQueue(优先队列),PriorityQueue内部使用完全二叉堆(不知道的自行了解哈)来实现队列元素排序,我们在向DelayQueue队列中添加元素时,会给元素一个Delay(延迟时间)作为排序条件,队列中最小的元素会优先放在队首。队列中的元素只有到了Delay时间才允许从队列中取出。队列中可以放基本数据类型或自定义实体类,在存放基本数据类型时,优先队列中元素默认升序排列,自定义实体类就需要我们根据类属性值比较计算了。
package com.schdule.util;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
class MyDelayedTask implements Delayed {
private long delayTime;
public MyDelayedTask(long delayTime) {
this.delayTime = delayTime;
}
@Override
public long getDelay(TimeUnit unit) {
return unit.convert(delayTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
}
@Override
public int compareTo(Delayed other) {
if (this.delayTime < ((MyDelayedTask) other).delayTime) {
return -1;
} else if (this.delayTime > ((MyDelayedTask) other).delayTime) {
return 1;
} else {
return 0;
}
}
public static void main(String[] args) throws InterruptedException {
DelayQueue<MyDelayedTask> queue = new DelayQueue<>();
queue.put(new MyDelayedTask(System.currentTimeMillis() + 5000)); // 5秒后执行
while (true) {
MyDelayedTask task = queue.take();
System.out.println("Executing task at: " + System.currentTimeMillis());
}
}
}
DelayQueue的put方法是线程安全的,因为put方法内部使用了ReentrantLock锁进行线程同步。DelayQueue还提供了两种出队的方法 poll() 和 take() , poll() 为非阻塞获取,没有到期的元素直接返回null;take() 阻塞方式获取,没有到期的元素线程将会等待。
Redis实现延迟队列
Redis的数据结构Zset,同样可以实现延迟队列的效果,主要利用它的score属性,redis通过score来为集合中的成员进行从小到大的排序。
Redis有序集合的一些常用命令包括:
ZADD: 向有序集合添加一个或多个成员,或者更新已存在成员的分数。
ZRANGE: 返回有序集合中指定区间内的成员列表。
ZREVRANGE: 返回有序集合中指定区间内的成员列表,按分数从高到低排序。
ZRANGEBYSCORE: 返回有序集合中指定分数区间内的成员列表。
ZREM: 移除有序集合中的一个或多个成员。
ZCARD: 获取有序集合的成员数。
ZSCORE: 获取有序集合中成员的分数。
/**
* 消费消息
*/
public void pollOrderQueue() {
while (true) {
Set<Tuple> set = jedis.zrangeWithScores(DELAY_QUEUE, 0, 0);
String value = ((Tuple) set.toArray()[0]).getElement();
int score = (int) ((Tuple) set.toArray()[0]).getScore();
Calendar cal = Calendar.getInstance();
int nowSecond = (int) (cal.getTimeInMillis() / 1000);
if (nowSecond >= score) {
jedis.zrem(DELAY_QUEUE, value);
System.out.println(sdf.format(new Date()) + " removed key:" + value);
}
if (jedis.zcard(DELAY_QUEUE) <= 0) {
System.out.println(sdf.format(new Date()) + " zset empty ");
return;
}
Thread.sleep(1000);
}
}
RabbitMQ 延时队列
利用 RabbitMQ 做延时队列是比较常见的一种方式,而实际上RabbitMQ 自身并没有直接支持提供延迟队列功能,而是通过 RabbitMQ 消息队列的 TTL和 DXL这两个属性间接实现的。
先来认识一下 TTL和 DXL两个概念:
Time To Live(TTL) :
TTL 顾名思义:指的是消息的存活时间,RabbitMQ可以通过x-message-tt参数来设置指定Queue(队列)和 Message(消息)上消息的存活时间,它的值是一个非负整数,单位为微秒。
RabbitMQ 可以从两种维度设置消息过期时间,分别是队列和消息本身
设置队列过期时间,那么队列中所有消息都具有相同的过期时间。
设置消息过期时间,对队列中的某一条消息设置过期时间,每条消息TTL都可以不同。
如果同时设置队列和队列中消息的TTL,则TTL值以两者中较小的值为准。而队列中的消息存在队列中的时间,一旦超过TTL过期时间则成为Dead Letter(死信)。
Dead Letter Exchanges(DLX)
DLX即死信交换机,绑定在死信交换机上的即死信队列。RabbitMQ的 Queue(队列)可以配置两个参数x-dead-letter-exchange 和 x-dead-letter-routing-key(可选),一旦队列内出现了Dead Letter(死信),则按照这两个参数可以将消息重新路由到另一个Exchange(交换机),让消息重新被消费。
x-dead-letter-exchange:队列中出现Dead Letter后将Dead Letter重新路由转发到指定 exchange(交换机)。
x-dead-letter-routing-key:指定routing-key发送,一般为要指定转发的队列。
队列出现Dead Letter的情况有:
消息或者队列的TTL过期
队列达到最大长度
消息被消费端拒绝(basic.reject or basic.nack)
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;
@Configuration
public class RabbitMQConfig {
@Bean
public Queue myQueue() {
Map<String, Object> args = new HashMap<>();
args.put("x-message-ttl", 3000); // 设置队列的过期时间为 3 秒
args.put("x-dead-letter-exchange", "myExchange"); // 设置死信交换机
args.put("x-dead-letter-routing-key", "myRoutingKey"); // 设置死信路由键
return new Queue("myQueue", false, false, false, args);
}
@Bean
public DirectExchange myExchange() {
return new DirectExchange("myExchange");
}
@Bean
public Binding binding(Queue myQueue, DirectExchange myExchange) {
return BindingBuilder.bind(myQueue).to(myExchange).with("myRoutingKey");
}
}