为何调用Java DelayQueue的take()方法不会阻塞整个数据结构？

问题背景 #

我正在研究java.util.concurrent.DelayQueue在多线程环境下的工作机制，发现它内部使用ReentrantLock，且每个方法开头都会获取该锁。我通过poll()/take()方法实现了生产者-消费者模式，测试代码如下：

public class DelayQueueTest {
    static DelayQueue<Delayed> delayQueue = new DelayQueue<>();
    public static final long executionTime = 2_000L;
    public static long timeStart;

    public static void go() throws InterruptedException {
        timeStart = System.currentTimeMillis();
        Thread consumer1 = new Thread(new Consumer());
        Thread producer = new Thread(new Producer());


        consumer1.start();
        Thread.sleep(200);
        producer.start();

        producer.join();
        consumer1.join();
    }

    static class Consumer implements Runnable {
        static DelayQueue<Delayed> delayQueue = DelayQueueTest.delayQueue;

        @Override
        public void run() {
            while (System.currentTimeMillis() - timeStart < executionTime) {
                System.out.println(String.format(
                        "Thread %s started taking at %d",
                        Thread.currentThread().getId(),
                        System.currentTimeMillis() - timeStart)
                );
                Delayed result;
                try {
                    result = delayQueue.take();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                System.out.print(String.format(
                        "Thread %s finished taking at %d... ",
                        Thread.currentThread().getId(),
                        System.currentTimeMillis() - timeStart)
                );

                if (result != null) System.out.println(String.format("with result %s", result.toString()));
                else System.out.println();
                System.out.format("Queue size: %d\n", delayQueue.size());
                try {
                    Thread.sleep(500);
                }
                catch (InterruptedException ex) {
                    System.out.println(ex);
                }
            }
        }
    }

    static class Producer implements Runnable {
        static DelayQueue<Delayed> delayQueue = DelayQueueTest.delayQueue;

        @Override
        public void run() {
            while (System.currentTimeMillis() - timeStart < executionTime) {
                System.out.println(String.format(
                        "Thread %s started offering at %d. Queue size: %d",
                        Thread.currentThread().getId(),
                        System.currentTimeMillis() - timeStart,
                        delayQueue.size())
                );
                delayQueue.offer(new DelayObject(300));
                System.out.println(String.format(
                        "Thread %s finished offering at %d. Queue size: %d",
                        Thread.currentThread().getId(),
                        System.currentTimeMillis() - timeStart,
                        delayQueue.size())
                );
                try {
                    Thread.sleep(300);
                }
                catch (InterruptedException ex) {
                    System.out.println(ex);
                }
            }
        }
    }
}

当消费者线程调用take()，200ms后生产者线程调用offer()时，我原本认为take()获取锁后会阻塞整个数据结构，offer()因无法获取锁而无法添加元素，甚至会出现死锁，但实际程序输出如下：

Thread 15 started taking at 2
Thread 16 started offering at 214. Queue size: 0
Thread 16 finished offering at 217. Queue size: 1
Thread 16 started offering at 527. Queue size: 1
Thread 16 finished offering at 527. Queue size: 1
Thread 15 finished taking at 527... with result org.example.DelayQueueTest$DelayObject@7148b5a8
Queue size: 1
Thread 16 started offering at 828. Queue size: 1
Thread 16 finished offering at 828. Queue size: 2
Thread 15 started taking at 1035
Thread 15 finished taking at 1035... with result org.example.DelayQueueTest$DelayObject@21819116
Queue size: 1
Thread 16 started offering at 1143. Queue size: 1
Thread 16 finished offering at 1143. Queue size: 2
Thread 16 started offering at 1453. Queue size: 2
Thread 16 finished offering at 1453. Queue size: 3
Thread 15 started taking at 1548
Thread 15 finished taking at 1548... with result org.example.DelayQueueTest$DelayObject@6bf427cb
Queue size: 2
Thread 16 started offering at 1769. Queue size: 2
Thread 16 finished offering at 1769. Queue size: 3

疑问 #

1. 为何会出现这样的输出？为何已被锁定的情况下offer()仍能成功执行？
2. 能否提供DelayQueue的潜在使用场景？

解答 #

1. take()与offer()的锁机制说明 #

你误解了DelayQueue中take()方法的锁行为：take()获取内部的ReentrantLock后，如果队列中没有到期的元素，它会释放锁并进入等待状态，而不是一直持有锁阻塞整个队列。

具体流程是：

• 当消费者调用take()时，先获取锁，检查队列头部元素是否到期：
  • 如果没有到期的元素，take()会调用Condition.await()方法，这个方法会自动释放当前持有的锁，让当前线程进入等待队列，直到有生产者添加元素或者等待超时。
  • 当生产者调用offer()时，它可以顺利获取锁，添加元素后，会调用Condition.signal()唤醒等待中的消费者线程，之后释放锁。
  • 被唤醒的消费者线程会重新获取锁，再次检查是否有到期元素，若有则取出返回，否则继续等待。

这就是为什么测试中offer()能成功执行——因为take()在等待时释放了锁，并没有一直占用。ReentrantLock配合Condition实现了这种“等待-唤醒”的协作模式，既保证了线程安全，又避免了锁的长期占用导致的阻塞。

另外，测试中第一个offer()添加的元素延迟是300ms，消费者在2ms时开始take()，此时元素还没到期，所以消费者进入等待，直到527ms时元素到期，消费者才被唤醒（或主动检查到到期），完成take()操作，这也对应了输出中线程15在527ms才完成take的结果。

2. DelayQueue的常见使用场景 #

• 定时任务调度：实现轻量级定时任务，无需依赖Quartz等框架时，可将带执行时间的任务对象存入DelayQueue，消费者线程不断take()到期任务并执行。
• 缓存失效处理：存储缓存条目，每个条目携带过期时间，消费者线程定期清理过期缓存项。
• 消息延迟推送：消息系统中，需延迟发送的消息放入DelayQueue，到期后再推送给接收方。
• 订单超时处理：电商场景中，未支付订单存入DelayQueue，到期后自动取消订单、释放库存。
• 连接超时管理：监控网络连接，将连接放入DelayQueue，到期后检查连接状态，关闭超时空闲连接。

内容的提问来源于stack exchange，提问作者Fedorucho