Java阻塞延迟队列DelayQueue原理及使用_Hopefully Sky的博客

阻塞队列

阻塞队列（BlockingQueue）是一个支持两个附加操作的队列。这两个附加的操作是：在队列为空时，从队列中获取元素的消费者线程会一直等待直到队列变为非空。当队列满时，向队列中放置元素的生产者线程会等待直到队列可用。阻塞队列常用于生产者和消费者的场景，生产者是往队列里添加元素的线程，消费者是从队列里拿元素的线程。

在阻塞队列不可用时，这两个附加操作提供了4种处理方式：

• 抛出异常：当队列满时，插入元素会抛出IllegalStateException；
• 返回特殊值：offer()是入队方法，当插入成功时返回true，插入失败返回false；poll()是出队方法，当出队成功时返回元素的值，队列为空时返回null
• 一直阻塞：当队列满时，阻塞执行插入方法的线程；当队列空时，阻塞执行出队方法的线程
• 超时退出：顾名思义

下面是Java常见的阻塞队列。

• ArrayBlockingQueue ：一个由数组结构组成的有界阻塞队列
• LinkedBlockingQueue ：一个由链表结构组成的有界阻塞队列
• PriorityBlockingQueue ：一个支持优先级排序的无界阻塞队列
• DelayQueue：一个使用优先级队列实现的无界阻塞队列
• SynchronousQueue：一个不存储元素的阻塞队列
• LinkedTransferQueue：一个由链表结构组成的无界阻塞队列
• LinkedBlockingDeque：一个由链表结构组成的双向阻塞队列

DelayQueue解析

DelayQueue队列中每个元素都有个过期时间，并且队列是个优先级队列，当从队列获取元素时候，只有过期元素才会出队，DelayQueue的类结构如下图所示：

如图DelayQueue中内部使用的是PriorityQueue存放数据，使用ReentrantLock实现线程同步，可知是阻塞队列。另外队列里面的元素要实现Delayed接口，一个是获取当前剩余时间的接口，一个是元素比较的接口，因为这个是有优先级的队列。

DelayQueue 的主要成员

public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
    implements BlockingQueue<E> {
    // 持有内部重入锁。
    private final transient ReentrantLock lock = new ReentrantLock();
    // 优先级队列，存放工作任务。
    private final PriorityQueue<E> q = new PriorityQueue<E>();
    private Thread leader = null;
    // 依赖于重入锁的condition。
    private final Condition available = lock.newCondition();
}

元素入队列

插入元素到队列中主要三个方法，但实际上底层调用的都是offer(e)方法

/**
 * Inserts the specified element into this delay queue.
 *
 * @param e the element to add
 * @return {@code true} (as specified by {@link Collection#add})
 * @throws NullPointerException if the specified element is null
 */
public boolean add(E e) {
    return offer(e);
}
/**
 * Inserts the specified element into this delay queue. As the queue is
 * unbounded this method will never block.
 *
 * @param e the element to add
 * @throws NullPointerException {@inheritDoc}
 */
public void put(E e) {
    offer(e);
}

/**
 * Inserts the specified element into this delay queue.
 *
 * @param e the element to add
 * @return {@code true}
 * @throws NullPointerException if the specified element is null
 */
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();
    }
}

首先获取独占锁，然后添加元素到优先级队列，由于q是优先级队列，所以添加完元素后，peek()方法返回的并不一定是刚才添加的元素，如果判断为true，说明当前元素e的优先级最小也就是即将过期的，这时候激活avaliable变量条件队列里面的线程，通知它们队列里面有元素了。

从队列中取元素

有两个方法可以取元素（都是取队头），poll()方法取队头当队头元素没过期时返回null，take()方法取队头当队头元素没过期时会一直等待。

/**
 * Retrieves and removes the head of this queue, or returns {@code null}
 * if this queue has no elements with an expired delay.
 *
 * @return the head of this queue, or {@code null} if this
 *         queue has no elements with an expired delay
 */
public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        E first = q.peek();
        //如果队列为空，或者不为空但是队头元素没有过期则返回null
        if (first == null || first.getDelay(NANOSECONDS) > 0)
            return null;
        else
            return q.poll();
    } finally {
        lock.unlock();
    }
}

/**
 * Retrieves and removes the head of this queue, waiting if necessary
 * until an element with an expired delay is available on this queue.
 *
 * @return the head of this queue
 * @throws InterruptedException {@inheritDoc}
 */
public E take() throws InterruptedException {
    // 获取锁。每个延迟队列内聚了一个重入锁。
    final ReentrantLock lock = this.lock;
    // 获取可中断的锁。
    lock.lockInterruptibly();
    try {
        for (;;) {
            // 尝试从优先级队列中获取队列头部元素，获取但不移除
            E first = q.peek();
            if (first == null)
                //无元素，当前线程节点加入等待队列，并阻塞当前线程
                available.await();
            else {
                // 通过延迟任务的getDelay()方法获取延迟时间
                long delay = first.getDelay(NANOSECONDS);
                if (delay <= 0)
                    //延迟时间到期，获取并删除头部元素。
                    return q.poll();
                first = null; // don't retain ref while waiting
                if (leader != null)
                    available.await();
                else {
                    Thread thisThread = Thread.currentThread();
                    leader = thisThread;
                    try {
                        // 线程节点进入等待队列 x 纳秒。
                        available.awaitNanos(delay);
                    } finally {
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }
    } finally {
        // 若还存在元素的话，则将等待队列头节点中的线程节点移动到同步队列中。
        if (leader == null && q.peek() != null)
            available.signal();
        lock.unlock();
    }
}

重点说一下take()方法，第一次调用take时候由于队列空，所以把当前线程放入available的条件队列等待，当执行offer()成功并且添加的新元素恰好就是优先级队列的队首时就会通知最先等待的线程激活，循环重新获取队首元素，这时候first假如不空，则调用getDelay()方法看该元素还剩下多少时间就过期了，如果delay<=0则说明已经过期，则直接出队返回。否则看leader是否为null，不为null则说明是其他线程也在执行take()则把当前线程放入条件队列，否则就是只有当前线程执行的take()方法，则当前线程await直到剩余过期时间到，这期间该线程会释放锁，所以其他线程可以offer()添加元素，也可以take()阻塞自己，剩余过期时间到后，当前线程会重新竞争锁，重新进入循环。

如果已经具备了JUC包中的Lock接口以及AQS的相关知识，上述代码大部分应该都比较容易理解。DelayQueue将实现了Delayed接口的对象添加到优先级队列中，通过在重入锁的Condition上调用 await() 方法，实现了延迟获取阻塞队列中元素的功能。对于可重入锁、Condition接口及AQS相关知识，可以参考这两篇博文：从源码角度理解ReentrantLock及队列同步器以及又见队列同步器——Condition接口的实现

总结

1. DelayQueue是一个内部依靠AQS队列同步器所实现的无界延迟阻塞队列。
2. 延迟对象需要覆盖 getDelay()与compareTo()方法，并且要注意 getDelay()的时间单位的统一，compareTo()根据业务逻辑进行合理的比较逻辑重写。
3. DelayQueue中内聚的重入锁是非公平的。
4. DelayQueue是实现定时任务的关键，ScheduledThreadPoolExecutor中就用到了DelayQueue。

DelayQueue使用例子

写一个简单的例子：

public class DelayQueueTest {

    public static final int SIZE = 10;

    public static void main(String[] args) {
    	DelayQueueTest test = new DelayQueueTest();
        //初始化线程池
        BlockingQueue<Runnable> arrayBlockingQueue = new ArrayBlockingQueue<>(10);
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor
            (5, 10, 10, TimeUnit.MILLISECONDS,
                arrayBlockingQueue, Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());

        DelayQueue<DelayedTask> delayTaskQueue = new DelayQueue<>();
        //模拟SIZE个延迟任务
        for (byte i = 0; i < SIZE; i++) {
            Long runAt = System.currentTimeMillis() + 1000 * i;
            String name = "Zhang_" + i;
            byte age = (byte)(10 + i);
            String gender = (i % 2 == 0 ? "male" : "female");
            Student student = new StudentBuilder(name, age, gender).height(150 + i).province("ZheJiang").build();
            delayTaskQueue.put(new DelayedTask<Student>(student, 1, function -> test.print(student), runAt));
        }

        while (true) {
            if (delayTaskQueue.size() == 0) {
                break;
            }
            try {
                //从延迟队列中取值,如果没有对象过期则取到null
                DelayedTask delayedTask = delayTaskQueue.poll();
                if (delayedTask != null) {
                    threadPool.execute(delayedTask);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        threadPool.shutdown();
    }


    public String print(Object object) {
      System.out.println(Thread.currentThread().getName());
        String str = ">>>junit log>>>" + object.getClass().getSimpleName() + ":" + object.toString();
        System.out.println(str);
        return str;
    }

    private static class DelayedTask<T> implements Delayed, Runnable {

        /**
         * 任务参数
         */
        private T taskParam;

        /**
         * 任务类型
         */
        private Integer type;

        /**
         * 任务函数
         */
        private Function<T, String> function;

        /**
         * 任务执行时刻
         */
        private Long runAt;

        public T getTaskParam() {
            return taskParam;
        }
        public Integer getType() {
            return type;
        }
        public Function<T, String> getFunction() {
            return function;
        }
        public Long getRunAt() {
            return runAt;
        }
        DelayedTask(T taskParam, Integer type, Function<T, String> function, Long runAt) {
            this.taskParam = taskParam;
            this.type = type;
            this.function = function;
            this.runAt = runAt;
        }
        @Override
        public void run() {
            if (taskParam != null) {
                function.apply(taskParam);
            }
        }
        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(this.runAt - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }
        @Override
        public int compareTo(Delayed o) {
            DelayedTask object = (DelayedTask)o;
            return this.runAt.compareTo(object.getRunAt());
        }
    }
}

运行结果如下，由于10个元素的延迟时间均相差1秒，可以看到逐步打印的效果。

DelayQueue典型场景是重试机制实现，比如当调用接口失败后，把当前调用信息放入delay=10s的元素，然后把元素放入队列，那么这个队列就是一个重试队列，一个线程通过take()方法获取需要重试的接口，take()返回则接口进行重试，失败则再次放入队列，同时也可以在元素加上重试次数。