网站建设哪里好薇,成都手机网站建设开发,网站开发+搜索,济南网站建设(选聚搜网络)文章目录 前言正文一、基本概念1.1 延时队列的特点1.2 常见的实现方式 二、Java原生的内存型延时队列2.1 定义延时元素DelayedElement2.2 定义延时队列管理器DelayedQueueManager2.3 消费元素2.4 调试2.5 调试结果2.6 精髓之 DelayQueue.poll() 三、基于Redisson的延时队列3.1 … 文章目录 前言正文一、基本概念1.1 延时队列的特点1.2 常见的实现方式 二、Java原生的内存型延时队列2.1 定义延时元素DelayedElement2.2 定义延时队列管理器DelayedQueueManager2.3 消费元素2.4 调试2.5 调试结果2.6 精髓之 DelayQueue.poll() 三、基于Redisson的延时队列3.1 定义延时队列管理器3.2 调试3.3 调试结果 前言
业务中经常会出现各种涉及到定时延迟执行的需求任务。
有一种队列专门处理这种情况。那就是延时队列。 本文提供两种实现方式 java原生的内存型延时队列redisson 的内置延时队列 正文
一、基本概念
延时队列Delay Queue是一种特殊的消息队列用于处理需要在将来某个时间点执行的任务。
与普通的队列不同延时队列中的消息在指定的时间之前是不可见的只有当消息的延时时间到达后消息才会被消费。
1.1 延时队列的特点
延时性消息在进入队列后并不会立即被消费而是需要等待一段时间后才能被消费。有序性消息按照延时时间的先后顺序被消费。可靠性通常需要保证消息不丢失即使在系统故障的情况下也能恢复。这里对于内存型的延时队列不太适合一旦内存释放就会丢失消息
1.2 常见的实现方式
数据库使用数据库的定时任务或触发器。消息队列使用支持延时消息的消息队列如 RabbitMQ、Kafka、RocketMQ 等。内存队列使用内存中的数据结构如 Java 中的 DelayQueue。Redis使用 Redis 的 sorted set 或 Redisson 的 RDelayedQueue。
二、Java原生的内存型延时队列
使用 Java 的 DelayQueue 生产者将任务封装成 Delayed 接口的实现类添加到 DelayQueue 中。消费者使用 take 或 poll 方法从 DelayQueue 中取出任务进行处理。 2.1 定义延时元素DelayedElement
package com.pine.common.util.delayqueue;import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;/*** 延迟元素** author fengjinsong*/
public class DelayedElement implements Delayed {/*** 延迟时间单位毫秒*/private final AtomicLong delayTime;/*** 到期时间*/private final AtomicLong expire;/*** 任务数据*/private final Object data;/*** 执行次数*/private final AtomicInteger executionFrequency;public DelayedElement(long delayTime, Object data) {this.delayTime new AtomicLong(delayTime);this.expire new AtomicLong(System.currentTimeMillis() delayTime);this.data data;this.executionFrequency new AtomicInteger(0);}public Object getData() {return this.data;}public AtomicInteger getExecutionFrequency() {return executionFrequency;}public void setExecutionFrequency() {this.executionFrequency.incrementAndGet();}Overridepublic long getDelay(TimeUnit unit) {return unit.convert(this.expire.longValue() - System.currentTimeMillis(), TimeUnit.MILLISECONDS);}Overridepublic int compareTo(Delayed o) {return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));}/*** 重置延迟时间*/public void resetDelay(long delayTime) {this.delayTime.set(delayTime);this.expire.set(System.currentTimeMillis() this.delayTime.longValue());}/*** 重置延迟时间*/public void resetDelay() {resetDelay(this.delayTime.longValue());}Overridepublic String toString() {return DelayedElement{ delayTime delayTime , expire expire , data data , executionFrequency executionFrequency };}
}
2.2 定义延时队列管理器DelayedQueueManager
package com.pine.common.util.delayqueue;import java.util.List;
import java.util.concurrent.DelayQueue;/*** 延时队列管理器** author fengjinsong*/
public class DelayedQueueManager {private DelayedQueueManager() {}/*** 延时队列*/private static final DelayQueueDelayedElement DELAY_QUEUE new DelayQueue();/*** 添加元素** param element 元素*/public static void addElement(DelayedElement element) {DELAY_QUEUE.add(element);}public static void addElement(ListDelayedElement elements) {DELAY_QUEUE.addAll(elements);}/*** 获取元素并从队列中移除该元素** return 元素*/public static DelayedElement pollElement() {return DELAY_QUEUE.poll();}
}2.3 消费元素
package com.pine.common.util.delayqueue;import java.time.LocalDateTime;public class DelayedElementConsumer implements Runnable {private final static int[] FREQUENCY_SEQUENCE new int[]{1, 2, 3, 6, 12, 24, 48, 96, 192, 384, 768};Overridepublic void run() {boolean hasDelayedElement true;while (hasDelayedElement) {// 获取元素DelayedElement element DelayedQueueManager.pollElement();try {if (element ! null) {System.out.println(LocalDateTime.now() 消费了延迟元素 element);if (element.getData().toString().contains(3)) {throw new RuntimeException(模拟报错);}} else {hasDelayedElement false;}} catch (Exception e) {retry(element);}}}private void retry(DelayedElement element) {element.setExecutionFrequency();System.out.println(执行出错 element);//出错3次后不再重试if (element.getExecutionFrequency().intValue() 3) {System.out.println(出错3次后不再重试);} else {element.resetDelay(FREQUENCY_SEQUENCE[element.getExecutionFrequency().intValue() 3] * 1000);// 重试DelayedQueueManager.addElement(element);}}}2.4 调试
package com.pine.common.redis.delayqueue;import java.time.LocalDateTime;
import java.util.concurrent.TimeUnit;public class Client {public static void main(String[] args) {// 模拟生产数据RedissonDelayedQueueManager.offer(hello22, 3000);RedissonDelayedQueueManager.offer(hello33, 5000);// 模拟消费数据System.out.println(LocalDateTime.now() 开始消费数据);while (true) {Object object RedissonDelayedQueueManager.poll(10, TimeUnit.SECONDS);if (object ! null) {System.out.println(----------------------- LocalDateTime.now() : object);}}}
}2.5 调试结果
2024-11-06T16:57:39.342358开始消费数据
-----------------------2024-11-06T16:57:42.285383:hello22
-----------------------2024-11-06T16:57:44.378298:hello33可以观察到 hello22 延时了3秒hello33延时了5秒
2.6 精髓之 DelayQueue.poll()
检索并删除此队列的头部如果此队列没有延迟过期的元素则返回null。
public E poll() {final ReentrantLock lock this.lock;lock.lock();try {E first q.peek();return (first null || first.getDelay(NANOSECONDS) 0)? null: q.poll();} finally {lock.unlock();}}三、基于Redisson的延时队列
使用 Redisson 的 RDelayedQueue 生产者使用 RDelayedQueue 的 offer 方法将任务添加到队列中指定延时时间。消费者使用 RQueue 的 poll 方法从队列中取出任务进行处理。 3.1 定义延时队列管理器
package com.pine.common.redis.delayqueue;import org.redisson.Redisson;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;import java.io.IOException;
import java.util.concurrent.TimeUnit;public class RedissonDelayedQueueManager {private static final String QUEUE_NAME delay_queue;private static final RedissonClient REDISSON_CLIENT;static {try {String content singleServerConfig:address: redis://10.189.64.136:8379;Config config Config.fromYAML(content);REDISSON_CLIENT Redisson.create(config);} catch (IOException e) {throw new RuntimeException(e);}}/*** 获取延迟队列* p* 本方法通过Redisson客户端创建一个阻塞队列并基于该阻塞队列创建一个延迟队列* 延迟队列用于处理需要延迟执行的任务例如任务重试机制、任务调度等场景** param T 队列中元素的类型* return 返回一个延迟队列实例用于后续的操作和管理*/private static T RDelayedQueueT getDelayedQueue() {// 创建一个阻塞队列这是后续创建延迟队列的基础RBlockingQueueT queue REDISSON_CLIENT.getBlockingQueue(QUEUE_NAME);// 基于阻塞队列创建延迟队列并返回return REDISSON_CLIENT.getDelayedQueue(queue);}/*** 向延迟队列中添加元素并设置延迟时间** param task 要添加的元素* param delayTime 延迟时间单位为毫秒* param T 元素类型*/public static T void offer(T task, long delayTime) {RDelayedQueueT delayedQueue getDelayedQueue();delayedQueue.offer(task, delayTime, TimeUnit.MILLISECONDS);}/*** 从延迟队列中获取元素并设置超时时间** param timeout 超时时间单位为毫秒* param unit 超时时间单位* param T 元素类型* return 返回获取到的元素如果没有获取到元素则返回null*/public static T T poll(long timeout, TimeUnit unit) {RBlockingQueueT queue REDISSON_CLIENT.getBlockingQueue(QUEUE_NAME);try {return queue.poll(timeout, unit);} catch (InterruptedException e) {throw new RuntimeException(e);}}
}
3.2 调试
package com.pine.common.redis.delayqueue;import java.time.LocalDateTime;
import java.util.concurrent.TimeUnit;public class Client {public static void main(String[] args) {// 模拟生产数据RedissonDelayedQueueManager.offer(hello22, 3000);RedissonDelayedQueueManager.offer(hello33, 5000);// 模拟消费数据System.out.println(LocalDateTime.now() 开始消费数据);while (true) {Object object RedissonDelayedQueueManager.poll(10, TimeUnit.SECONDS);if (object ! null) {System.out.println(----------------------- LocalDateTime.now() : object);}}}
}
3.3 调试结果
2024-11-06T17:05:31.630768开始消费数据
-----------------------2024-11-06T17:05:34.548032:hello22
-----------------------2024-11-06T17:05:36.732607:hello33可以观察到 hello22 延时了3秒hello33延时了5秒
