public class Order implements Delayed { /** * 堆代码 duidaima.com * 延迟时间 */ @JsonFormat(locale = "zh", timezone = "GMT+8", pattern = "yyyy-MM-dd HH:mm:ss") private long time; String name; public Order(String name, long time, TimeUnit unit) { this.name = name; this.time = System.currentTimeMillis() + (time > 0 ? unit.toMillis(time) : 0); } @Override public long getDelay(TimeUnit unit) { return time - System.currentTimeMillis(); } @Override public int compareTo(Delayed o) { Order Order = (Order) o; long diff = this.time - Order.time; if (diff <= 0) { return -1; } else { return 1; } } }DelayQueue的put方法是线程安全的,因为put方法内部使用了ReentrantLock锁进行线程同步。DelayQueue还提供了两种出队的方法 poll() 和 take() , poll() 为非阻塞获取,没有到期的元素直接返回null;take() 阻塞方式获取,没有到期的元素线程将会等待。
public class DelayQueueDemo { public static void main(String[] args) throws InterruptedException { Order Order1 = new Order("Order1", 5, TimeUnit.SECONDS); Order Order2 = new Order("Order2", 10, TimeUnit.SECONDS); Order Order3 = new Order("Order3", 15, TimeUnit.SECONDS); DelayQueue<Order> delayQueue = new DelayQueue<>(); delayQueue.put(Order1); delayQueue.put(Order2); delayQueue.put(Order3); System.out.println("订单延迟队列开始时间:" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))); while (delayQueue.size() != 0) { /** * 取队列头部元素是否过期 */ Order task = delayQueue.poll(); if (task != null) { System.out.format("订单:{%s}被取消, 取消时间:{%s}\n", task.name, LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))); } Thread.sleep(1000); } } }上边只是简单的实现入队与出队的操作,实际开发中会有专门的线程,负责消息的入队与消费。
订单延迟队列开始时间:2020-05-06 14:59:09 订单:{Order1}被取消, 取消时间:{2020-05-06 14:59:14} 订单:{Order2}被取消, 取消时间:{2020-05-06 14:59:19} 订单:{Order3}被取消, 取消时间:{2020-05-06 14:59:24}2、Quartz 定时任务
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-quartz</artifactId> </dependency>在启动类中使用@EnableScheduling注解开启定时任务功能。
@EnableScheduling @SpringBootApplication public class DelayqueueApplication { public static void main(String[] args) { SpringApplication.run(DelayqueueApplication.class, args); } }编写一个定时任务,每个5秒执行一次。
@Component public class QuartzDemo { //每隔五秒 @Scheduled(cron = "0/5 * * * * ? ") public void process(){ System.out.println("我是定时任务!"); } }3、Redis sorted set
zadd delayqueue 3 order3消费端轮询队列delayqueue, 将元素排序后取最小时间与当前时间比对,如小于当前时间代表已经过期移除key。
/** * 消费消息 */ 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); } }我们看到执行结果符合预期
2020-05-07 13:24:09 add finished. 2020-05-07 13:24:19 removed key:order1 2020-05-07 13:24:29 removed key:order2 2020-05-07 13:24:39 removed key:order3 2020-05-07 13:24:39 zset empty4、Redis 过期回调
notify-keyspace-events ExRedis监听配置,注入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 并给定3s的过期时间。
set xiaofu 123 ex 3在控制台成功监听到了这个过期的key。
监听到过期的key为:xiaofu5、RabbitMQ 延时队列
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、时间轮
public HashedWheelTimer(ThreadFactory threadFactory, long tickDuration, TimeUnit unit, int ticksPerWheel) { this(threadFactory, tickDuration, unit, ticksPerWheel, true); }TimerTask:一个定时任务的实现接口,其中run方法包装了定时任务的逻辑。
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为定时任务,按照不同的周期重复执行。
order2 10s 后执行