石家庄住房和城乡建设局网站,帝国网站后台管理系统,为什么做网站特效用用插件,苏州北京商场网站建设JUC系列整体栏目 内容链接地址【一】深入理解JMM内存模型的底层实现原理https://zhenghuisheng.blog.csdn.net/article/details/132400429【二】深入理解CAS底层原理和基本使用https://blog.csdn.net/zhenghuishengq/article/details/132478786【三】熟练掌握Atomic原子系列基本…JUC系列整体栏目 内容链接地址【一】深入理解JMM内存模型的底层实现原理https://zhenghuisheng.blog.csdn.net/article/details/132400429【二】深入理解CAS底层原理和基本使用https://blog.csdn.net/zhenghuishengq/article/details/132478786【三】熟练掌握Atomic原子系列基本使用https://blog.csdn.net/zhenghuishengq/article/details/132543379【四】精通Synchronized底层的实现原理https://blog.csdn.net/zhenghuishengq/article/details/132740980【五】通过源码分析AQS和ReentrantLock的底层原理https://blog.csdn.net/zhenghuishengq/article/details/132857564【六】深入理解Semaphore底层原理和基本使用https://blog.csdn.net/zhenghuishengq/article/details/132908068【七】深入理解CountDownLatch底层原理和基本使用https://blog.csdn.net/zhenghuishengq/article/details/133343440【八】深入理解CyclicBarrier底层原理和基本使用https://blog.csdn.net/zhenghuishengq/article/details/133378623【九】深入理解ReentrantReadWriteLock 读写锁的底层实现https://blog.csdn.net/zhenghuishengq/article/details/133629550【十】深入理解ArrayBlockingQueue的基本使用和底层实现https://blog.csdn.net/zhenghuishengq/article/details/133692023【十一】深入理解LinkedBlockingQueue的基本使用和底层实现https://blog.csdn.net/zhenghuishengq/article/details/133723652【十二】深入理解PriorityQueue的基本使用和底层实现https://blog.csdn.net/zhenghuishengq/article/details/133788655【十三】深入理解DelayQueue的基本使用和底层实现https://blog.csdn.net/zhenghuishengq/article/details/133820599 深入理解DelayQueue延迟队列的底层原理 一深入理解DelayQueue延迟队列1DelayQueue的基本使用2DelayQueue的底层源码分析2.1DelayQueue类属性2.2入队offer方法2.3出队take方法 3总结 一深入理解DelayQueue延迟队列
延时队列顾名思义就是可以实现在一段时间之后在执行这个任务。在分布式场景下可能会更加的选择使用MQ来完成这些操作但是在单JVM进程中或者在mq挂了的兜底方案中会考虑使用这个DelayQueue来完成这个延时任务的。如一些订单超时未支付任务超时管理短信异步通知等情况就可以使用这个延时队列来完成了。
在了解这个DelayQueue延迟队列之前需要先熟悉上一篇PriorityQueue的基本使用和底层原理因为这个延迟队列的底层的数据结构就是通过这个优先级队列来实现的
class DelayQueueE extends Delayed extends AbstractQueueE implements BlockingQueueE{//组合了一个优先级队列private final PriorityQueueE q new PriorityQueueE();
}由于这个优先级队列采用的是二叉堆的数据结构并且采用的是小顶堆的数据结构因此很容易猜出这个DelayQueue的底层原理了就是假设5个延时任务会将最近到期的这个任务排在阻塞队列的前面因此在出队的时候就可以保证先过期的先出队。
由于底层是通过这个PriorityQueue的优先级队列实现的因此这个DelayQueue也是一个无界的阻塞队列在使用这个延迟队列时需要实现一个Delayed 的接口。总而言之就是不保证先进先出下一个即将过期的任务会排到队列的最前面
1DelayQueue的基本使用
由于在实际开发中会有这种订单超时的场景因此这里主要是模拟一个订单的超时任务来体验一下这个DelayQueue的基本使用
首先创建一个实现了Delayed接口的OrderDelay订单延时类Delayed也是Comparable类的一个具体实现
/*** Delayed的具体的方法实现* Author: zhenghuisheng* Date: 2023/10/14 0:32*/
Data
public class OrderDelay implements Delayed {//需要延迟的时间private long delayTime;//订单idprivate Integer orderId;//商品名称private String productName;//构造方法public OrderDelay(long delayTime,Integer orderId,String productName){//需要延迟的 时间 当前系统的时间this.delayTime delayTime System.currentTimeMillis();this.orderId orderId;this.productName productName;}//获取剩余的延时时间Overridepublic long getDelay(TimeUnit unit) {//到达时间 - 剩余时间long residueTime this.delayTime - System.currentTimeMillis();return unit.convert(residueTime,TimeUnit.MILLISECONDS);}//实现这个比较器方法Overridepublic int compareTo(Delayed o) {OrderDelay orderDelay (OrderDelay)o;return orderDelay.delayTime this.delayTime ? - 1 : 1;}
}随后创建一个生产者Producer线程任务类用于将为支付的订单加入到这个延时队列中
Data
public class Producer implements Runnable {//全局的阻塞队列private DelayQueue queue;//延迟队列订单类对象private OrderDelay orderDelay;public Producer(DelayQueue queue,OrderDelay orderDelay){this.queue queue;this.orderDelay orderDelay;}//添加文件Overridepublic void run() {try {queue.put(orderDelay); //加入阻塞队列System.out.println(orderDelay.getProductName() 加入完毕...);} catch (Exception e) {e.printStackTrace();}}
}随后创建一个消费者Consumer线程任务类用于取出即将过期的订单任务
/*** 消费者线程* Author: zhenghuisheng* Date: 2023/10/8 20:21*/
Data
public class Consumer implements Runnable {private DelayQueue queue;public Consumer(DelayQueue delayQueue){this.queue delayQueue;}Overridepublic void run() {try {System.out.println(queue.take());} catch (InterruptedException e) {e.printStackTrace();}}
}然后再创建一个线程池的工具类用于更好的监控和管理线程
/*** 线程池工具* author zhenghuisheng* date : 2023/3/22*/
public class ThreadPoolUtil {/*** io密集型最大核心线程数为2N,可以给cpu更好的轮换* 核心线程数不超过2N即可可以适当留点空间* cpu密集型最大核心线程数为N或者N1N可以充分利用cpu资源N加1是为了防止缺页造成cpu空闲* 核心线程数不超过N1即可* 使用线程池的时机1,单个任务处理时间比较短 2,需要处理的任务数量很大*/public static synchronized ThreadPoolExecutor getThreadPool() {if (pool null) {//获取当前机器的cpuint cpuNum Runtime.getRuntime().availableProcessors();log.info(当前机器的cpu的个数为 cpuNum);int maximumPoolSize cpuNum * 2 ;pool new ThreadPoolExecutor(maximumPoolSize - 2,maximumPoolSize,5L, //5sTimeUnit.SECONDS,new LinkedBlockingQueue(), //数组有界队列Executors.defaultThreadFactory(), //默认的线程工厂new ThreadPoolExecutor.AbortPolicy()); //直接抛异常默认异常}return pool;}
}最后创建一个有Main方法的测试类用于对这个DelayQueue进行测试
/*** Author: zhenghuisheng* Date: 2023/10/14 1:41*/
public class DelayQueueDemo {//创建一个线程池static ThreadPoolExecutor pool ThreadPoolUtil.getThreadPool();//创建一个全局的延迟队列static DelayQueueOrderDelay delayQueue new DelayQueue();public static void main(String[] args) throws Exception {//生产者创建任务for (int i 7; i 2; i--) {OrderDelay orderDelay new OrderDelay(i * 1000, i, id_ i);//创建生产者线程Producer producerTask new Producer(delayQueue, orderDelay);//提交到线程池pool.execute(producerTask);}Thread.sleep(50);System.out.println(生产者线程创建完毕);//创建消费者线程for (int i 0; i 5; i) {Consumer consumerTask new Consumer(delayQueue);pool.execute(consumerTask);}}
}
最后看执行结果先进来但是延迟时间长所以后出去 id_7加入完毕… id_5加入完毕… id_6加入完毕… id_4加入完毕… id_3加入完毕… 生产者线程创建完毕 OrderDelay(delayTime1697221724133, orderId3, productNameid_3) OrderDelay(delayTime1697221725133, orderId4, productNameid_4) OrderDelay(delayTime1697221726132, orderId5, productNameid_5) OrderDelay(delayTime1697221727132, orderId6, productNameid_6) OrderDelay(delayTime1697221728129, orderId7, productNameid_7) 2DelayQueue的底层源码分析
2.1DelayQueue类属性
首先查看这个 DelayQueue 类也是继承了这个抽象类也是实现了这个BlockingQueue
class DelayQueueE extends Delayed extends AbstractQueueE implements BlockingQueueE在这个类中首先最重要的就是这个PriorityQueue优先级队列说明这个延迟队列的底层是通过这个优先级队列实现的
//组合了一个优先级队列
private final PriorityQueueE q new PriorityQueueE();随后就是一把互斥锁加一个条件队列组成互斥锁就是offer方法和take方法的互斥然后这个条件队列是在队列为空时存储这个线程结点的
private final transient ReentrantLock lock new ReentrantLock();
private final Condition available lock.newCondition();还有一个重要的属性就是一个leader的线程标记用对记录队头的线程谁最早过期就记录谁
private Thread leader null;最后来看看该类的构造方法里面是空的因为里面的offer和take主要是操作这个PriorityQueue类
public DelayQueue() {}2.2入队offer方法
接下来直接看这个类的offer方法的具体实现这下面的逻辑是比较简单的就是先入队如果是第一个元素入队那么回去唤醒条件队列中被阻塞的结点因为这些结点是队列为空而将线程阻塞的现在队列已经不为空了
public boolean offer(E e) {final ReentrantLock lock this.lock; //获取这把互斥锁lock.lock(); //加锁try {q.offer(e); //线程入队if (q.peek() e) { //如果堆顶为当前元素表示第一个元素入队leader null;available.signal(); //那么就会去唤醒因对列为空而被阻塞的线程结点}return true;} finally {lock.unlock(); //解锁}
}随后依旧是进入上面的offer方法做一个具体的入队操作这里需要结合PriorityQueue的属性来看首先会判断这个数组是否达到设置的最大值或者扩容后的最大值如果是则继续扩容
public boolean offer(E e) {if (e null) throw new NullPointerException(); //结点为空modCount; int i size;if (i queue.length) //达到最大值grow(i 1); //扩容size i 1; if (i 0) queue[0] e; //队列为空则直接加入堆顶else siftUp(i, e); //否则上浮堆算法return true;
}数组扩容的方法如下先做一个扩容操作并且最后创建一个新数组将旧值copy到新数组中随后将新数组返回假设此时的容量小于64则扩大原来的容量2如果大于64则扩大原来的容量一倍。
就是说假设此时容量为16那么第一次扩容就是 16162为34第二次扩容为34 34 2为70第三次扩容为70 70*2 210。
private void grow(int minCapacity) {int oldCapacity queue.length;// Double size if small; else grow by 50%int newCapacity oldCapacity ((oldCapacity 64) ?(oldCapacity 2) :(oldCapacity 1));// overflow-conscious codeif (newCapacity - MAX_ARRAY_SIZE 0)newCapacity hugeCapacity(minCapacity);queue Arrays.copyOf(queue, newCapacity);
}如果此时不是第一个结点入队那么就会调用这个 siftUp 方法如果有自定义实现的比较器则用自定义的否则则直接使用内部默认的比较器
private void siftUp(int k, E x) {if (comparator ! null)siftUpUsingComparator(k, x);elsesiftUpComparable(k, x);
}接着直接来看内部默认实现的这个上浮的方法吧就是一个小顶堆的入队操作
private static T void siftUpComparable(int k, T x, Object[] array) {Comparable? super T key (Comparable? super T) x; //创建一个比较构造器while (k 0) { //队列的元素值int parent (k - 1) 1; //获取当前结点的父节点的索引左移一位即可Object e array[parent]; //根据索引下标取值if (key.compareTo((T) e) 0) //比较和交换如果当前值大于父节点则不动break;array[k] e; //如果当前结点的值小于父结点则将当前结点改成父结点的值(默认使用的是小顶堆)k parent; //k在这个while循环下一定会等于0因此会走最下面的赋值就是不断地通过while循环将最小的交换到最上面}array[k] key; //如果队列的长度为0则直接将堆顶元素赋值
}在成功入队之后最后会调用这个unlock方法用于解锁并且唤醒被阻塞的结点
lock.unlock();2.3出队take方法
在结点入队之后那么接下来就看这个结点出队的方法出队方法相对来说是稍微多一点的。首先出队第一个头结点如果已经过期则直接出队否者获取这个即将过期的时间延迟阻塞即阻塞到到一定的时间主动唤醒最后执行这个任务会在这个for自旋中可以保证所有的结点出队。并且通过一个临时变量 leader只需获取最早过期的结点进行阻塞从而不需关心比该结点更晚过期的结点从而减少阻塞的数量。
public E take() throws InterruptedException {final ReentrantLock lock this.lock;lock.lockInterruptibly();try {for (;;) { //自旋E first q.peek(); //头结点出队if (first null) //如果队列为空available.await(); //则加入条件队列阻塞else {//获取头结点的过期时间long delay first.getDelay(NANOSECONDS); //头结点的过期时间小于0说明已经过期。则直接出队if (delay 0) return q.poll();first null; // dont retain ref while waiting//特别说明这个leader就是用于记录最早过期的那个线程if (leader ! null) //如果已经存在记录的最近过期的结点available.await(); //则阻塞else {Thread thisThread Thread.currentThread();leader thisThread;try {//延时阻塞阻塞到一定时间主动唤醒available.awaitNanos(delay);} finally {if (leader thisThread)leader null;}}}}} finally {if (leader null q.peek() ! null) available.signal(); //优化主动唤醒lock.unlock(); //解锁}
}最后会通过unlock进行一个解锁操作。
lock.unlock();3总结
延迟队列的底层是通过这个优先级队列来实现的越早过期的结点越先出队内部也是采用ReentrantLock条件队列来实现安全问题以及性能问题。延迟队列的结构也是无界队列形成的数组在入队的结点元素需要时Delayed类的具体实现。
