基本的网站建设知识,网站开发设计培训,企业网站开发,金泉网做网站要找谁文章目录 延时队列JDK自带的延时队列实现Redis实现延迟队列RabbitMQ 延时队列 延时队列
延时队列是一种特殊类型的队列#xff0c;它允许元素在特定时间间隔后才能被处理。这种队列在处理具有延迟需求的任务时非常有用#xff0c;例如定时任务、事件驱动系统等 延时队列在项… 文章目录 延时队列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;}Overridepublic long getDelay(TimeUnit unit) {return unit.convert(delayTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);}Overridepublic 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 {DelayQueueMyDelayedTask 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() 为非阻塞获取没有到期的元素直接返回nulltake() 阻塞方式获取没有到期的元素线程将会等待。
Redis实现延迟队列
Redis的数据结构Zset同样可以实现延迟队列的效果主要利用它的score属性redis通过score来为集合中的成员进行从小到大的排序。 Redis有序集合的一些常用命令包括 ZADD: 向有序集合添加一个或多个成员或者更新已存在成员的分数。 ZRANGE: 返回有序集合中指定区间内的成员列表。 ZREVRANGE: 返回有序集合中指定区间内的成员列表按分数从高到低排序。 ZRANGEBYSCORE: 返回有序集合中指定分数区间内的成员列表。 ZREM: 移除有序集合中的一个或多个成员。 ZCARD: 获取有序集合的成员数。 ZSCORE: 获取有序集合中成员的分数。 /*** 消费消息*/public void pollOrderQueue() {while (true) {SetTuple 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 ExchangesDLX 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 {Beanpublic Queue myQueue() {MapString, 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);}Beanpublic DirectExchange myExchange() {return new DirectExchange(myExchange);}Beanpublic Binding binding(Queue myQueue, DirectExchange myExchange) {return BindingBuilder.bind(myQueue).to(myExchange).with(myRoutingKey);}
}
