调用 demoimportjava.util.concurrent.DelayQueue;importjava.util.concurrent.Delayed;importjava.util.concurrent.TimeUnit;classTaskimplementsDelayed{privatelongexpire;privateStringname;publicTask(Stringname,longdelay,TimeUnitunit){this.namename;this.expireSystem.currentTimeMillis()unit.toMillis(delay);}OverridepubliclonggetDelay(TimeUnitunit){longremainexpire-System.currentTimeMillis();returnunit.convert(remain,TimeUnit.MILLISECONDS);}OverridepublicintcompareTo(Delayedo){returnLong.compare(this.getDelay(TimeUnit.MILLISECONDS),o.getDelay(TimeUnit.MILLISECONDS));}OverridepublicStringtoString(){returnname;}}publicclassDelayQueueMain{publicstaticvoidmain(String[]args)throwsInterruptedException{DelayQueueTaskqueuenewDelayQueue();queue.put(newTask(1s任务,1,TimeUnit.SECONDS));queue.put(newTask(3s任务,3,TimeUnit.SECONDS));queue.put(newTask(2s任务,2,TimeUnit.SECONDS));for(inti0;i3;i){Tasktqueue.take();// 取出到期1s任务// 取出到期2s任务// 取出到期3s任务System.out.println(取出到期t);}}}成员变量// 底层最小堆存储延迟元素privatefinalPriorityQueueEqnewPriorityQueueE();// 全局独占锁所有读写加锁privatefinalReentrantLocklocknewReentrantLock();// 等待条件无到期元素/队空时阻塞take线程privatefinalConditionavailablelock.newCondition();// Leader-Follower 优化标记privateThreadleadernull;构造方法// 无参构造器 main new 时执行 publicDelayQueue(){}// 带集合构造器main未使用附原版publicDelayQueue(Collection?extendsEc){this.addAll(c);}入队方法putpublicvoidput(Ee){offer(e);}addpublicbooleanadd(Ee){returnoffer(e);}offer超时publicbooleanoffer(Ee,longtimeout,TimeUnitunit){returnoffer(e);}offerpublicbooleanoffer(Ee){finalReentrantLocklockthis.lock;lock.lock();try{q.offer(e);// 放入堆自动上浮排序// 如果新元素是堆顶清空 leader 并唤醒 1 个线程if(q.peek()e){leadernull;available.signal();}returntrue;}finally{lock.unlock();}}出队方法takepublicEtake()throwsInterruptedException{finalReentrantLocklockthis.lock;lock.lockInterruptibly();// 可中断锁响应线程interrupttry{for(;;){// 死循环自旋校验队首Efirstq.peek();// 分支1队列为空无限阻塞等待if(firstnull)available.await();else{longdelayfirst.getDelay(NANOSECONDS);// 分支2队首已到期直接弹出返回if(delay0)returnq.poll();// 未到期释放引用防止内存泄漏firstnull;// 分支3已有 leader 线程当前线程永久阻塞if(leader!null)available.await();else{// 无leader当前线程成为 leader定时等待剩余延迟ThreadthisThreadThread.currentThread();leaderthisThread;try{available.awaitNanos(delay);}finally{// 等待结束清空leader标记if(leaderthisThread)leadernull;}}}}}finally{// 释放锁前如果无 leader 且队列有元素唤醒一个等待线程if(leadernullq.peek()!null)available.signal();lock.unlock();}}pollpublicEpoll(){finalReentrantLocklockthis.lock;lock.lock();try{Efirstq.peek();// 空 或 未到期 返回nullif(firstnull||first.getDelay(NANOSECONDS)0)returnnull;elsereturnq.poll();}finally{lock.unlock();}}总结无界阻塞队列put/offer 永不阻塞无容量限制底层最小堆队首一定是最快到期元素线程安全全局一把 ReentrantLock 所有操作都会上锁leader 等待优化减少多线程定时竞争提升并发性能仅到期可出队take/poll 不会返回未到期元素未到期线程阻塞相关考点first null 为什么可以防止内存泄漏不执行 first null 时的引用关系如下阻塞线程的局部变量 first 引用堆顶延迟任务对象 Delayed TaskDelayQueue 内部 PriorityQueue 也引用同一个 Task当线程执行到 available.await() / awaitNanos() 阻塞时线程挂起、长时间休眠线程栈帧不会销毁局部变量 first 会一直存活双重强引用导致 GC 无法回收队列里本来就有一份引用再加线程局部变量的强引用当这个任务到期被别的线程 poll 取出、从堆里删掉当前阻塞线程的变量 first 依旧死死持有该对象只要线程没被唤醒、栈帧没退出Task 及其关联大内存就无法被 GC长时间大量阻塞线程会造成内存泄漏底层组成存储PriorityQueue 无界最小堆依靠 Delayed 的 compareTo 排序锁全局一把 ReentrantLock 独占锁所有读写串行条件变量Condition available空队列 / 未到期时阻塞 take 线程优化标记Thread leaderLeader-Follower 等待模型元素强制约束存入元素必须实现 Delayed 接口getDelay(TimeUnit)判断是否到期的核心方法返回负数 已过期compareTo堆排序依据必须用剩余延迟时间比较写反会取出顺序错乱队列边界特性无界阻塞队列put/offer/add 永远不会阻塞不存在满队列remainingCapacity 固定返回 Integer.MAX_VALUEoffer 核心逻辑加锁 → 堆入队排序 → 判断新元素是否为堆顶if (q.peek() e)新任务是最快到期元素清空 leader、signal 唤醒等待线程不是堆顶则不唤醒减少无效系统唤醒take 核心阻塞流程整体死循环自旋逻辑获取可中断锁 lock.lockInterruptibly()取堆顶 peek()队列为空available.await() 无限阻塞有元素计算剩余 delaydelay 0直接poll()弹出返回未到期first null 切断引用防泄漏已有 leaderawait() 永久阻塞无 leader当前线程成为 leaderawaitNanos(delay) 限时等待finally 块释放锁前若无 leader 且队列有元素signal 唤醒一个线程take 中 finally 里为什么要执行 signal当前 leader 线程等待结束退出队列仍有未到期任务唤醒其他 follower 线程竞争新 leaderleader 线程的作用作用解决多线程同时 take 大量定时竞争减少系统唤醒、降低 CPU 自旋机制同一时刻仅允许 1 个 leader 线程执行awaitNanos(剩余延迟)其余 follower 线程全部await()无限阻塞。leader 到期唤醒后follower 才会竞争新 leader。如果没有 leader 机制会有什么问题上千线程同时 take全部执行 awaitNanos 定时等待每个任务到期全部线程同时唤醒竞争锁大量上下文切换、CPU 飙升。为什么新元素是堆顶才唤醒线程场景 1新插入元素变成新堆顶q.peek () e原有堆顶延迟更长新来的任务更快到期举例子原来堆顶是 5s 任务leader 线程等待 5s现在 put 一个 1s 任务堆顶变成 1s。此时存在两个问题旧 leader 还在等 5s但实际 1s 就有任务到期线程白白多等 4s无法及时消费旧 leader 记录的等待时长失效需要重置 leader null让下一次循环重新选出新 leader、等待新的 1s。所以必须leader null 清空旧 leader 标记available.signal() 唤醒阻塞线程重新走一次 take() 循环读取最新堆顶、重新计算等待时间。场景 2新插入元素不是堆顶堆顶不变堆顶还是原来那个最快到期的任务剩余延迟时间没有任何变化。leader 线程等待的时长依然准确不需要唤醒任何线程等到原有堆顶时间到自然唤醒即可如果此时无脑 signal()会造成无效唤醒线程被唤醒抢到锁后再次校验堆顶发现任务还没到期又会阻塞回去。频繁上下文切换拉高 CPU毫无收益。为什么只 signal() 而不是 signalAll()只需要唤醒一个线程重新竞争 leader唤醒全部会造成大量线程同时抢锁竞争加剧如果去掉这段判断每次 put 都执行 signal 会有什么后果每次入队都唤醒阻塞线程大量无效唤醒线程频繁阻塞 / 唤醒切换CPU 飙升并发性能严重下降。为什么要同步执行 leader null旧 leader 等待的是过期的长延迟标记作废唤醒后线程会重新竞争成为新 leader等待更短的延迟。