Java Review - 并发编程_DelayQueue原理&源码剖析

/**
 * A mix-in style interface for marking objects that should be
 * acted upon after a given delay.
 *
 * An implementation of this interface must define a
 * {@code compareTo} method that provides an ordering consistent with
 * its {@code getDelay} method.
 *
 * @since 1.5
 * @author Doug Lea
 */
public interface Delayed extends Comparable<Delayed> {

    /**
     * Returns the remaining delay associated with this object, in the
     * given time unit.
     *
     * @param unit the time unit
     * @return the remaining delay; zero or negative values indicate
     * that the delay has already elapsed
     */
    long getDelay(TimeUnit unit);
}

 private final transient ReentrantLock lock = new ReentrantLock();
 
  /**
     * Condition signalled when a newer element becomes available
     * at the head of the queue or a new thread may need to
     * become leader.
     */
    private final Condition available = lock.newCondition();

    /**
     * Thread designated to wait for the element at the head of
     * the queue.  This variant of the Leader-Follower pattern
     * (http://www.cs.wustl.edu/~schmidt/POSA/POSA2/) serves to
     * minimize unnecessary timed waiting.  When a thread becomes
     * the leader, it waits only for the next delay to elapse, but
     * other threads await indefinitely.  The leader thread must
     * signal some other thread before returning from take() or
     * poll(...), unless some other thread becomes leader in the
     * interim.  Whenever the head of the queue is replaced with
     * an element with an earlier expiration time, the leader
     * field is invalidated by being reset to null, and some
     * waiting thread, but not necessarily the current leader, is
     * signalled.  So waiting threads must be prepared to acquire
     * and lose leadership while waiting.
     */
    private Thread leader = null;

import java.util.Random;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

/**
 * @author 小工匠
 * @version 1.0
 * @description: TODO
 * @date 2021/12/19 23:05
 * @mark: show me the code , change the world
 */
public class DelayQueueTest {

    static class DelayedEle implements Delayed {

        private final long delayTime; //延迟时间
        private final long expire;  //到期时间
        private String data;   //数据

        public DelayedEle(long delay, String data) {
            delayTime = delay;
            this.data = data;
            expire = System.currentTimeMillis() + delay;
        }

        /**
         * 剩余时间=到期时间-当前时间
         */
        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(this.expire - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }

        /**
         * 优先队列里面优先级规则
         */
        @Override
        public int compareTo(Delayed o) {
            return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
        }

        @Override
        public String toString() {
            final StringBuilder sb = new StringBuilder("DelayedElement{");
            sb.append("delay=").append(delayTime);
            sb.append(", expire=").append(expire);
            sb.append(", data='").append(data).append('\'');
            sb.append('}');
            return sb.toString();
        }
    }


    public static void main(String[] args) throws InterruptedException {

        // 1 创建延时队列
        DelayQueue<DelayedEle> delayQueue = new DelayQueue<DelayedEle>();


        // 2 创建延时任务
        Random random = new Random();
        for (int i = 0; i < 10; i++) {
            DelayedEle ele = new DelayedEle(random.nextInt(500), "task-" + i);
            delayQueue.offer(ele);
        }


        System.out.println("开始操作，delayQueue队列大小为：" + delayQueue.size());


        // 3 依次取出任务并打印
        DelayedEle delayedEle = null;

        try {
            // 3.1 循环，如果想避免虚假唤醒，则不能把全部元素都打印出来
            for (; ; ) {
                // 3.2 获取过期的任务并打印
                while ((delayedEle = delayQueue.take()) != null) {
                    System.out.println(delayedEle.toString());
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }
}

    /**
     * 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; // 1 
        lock.lock();
        try {
            q.offer(e);
            if (q.peek() == e) {// 2 
                leader = null;
                available.signal();
            }
            return true;
        } 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 (;;) {
            	// 1 获取但不移除队首元素 
                E first = q.peek();
                if (first == null) 
                    available.await(); // 2 
                else {
                    long delay = first.getDelay(NANOSECONDS);
                    if (delay <= 0) // 3 
                        return q.poll();
                    first = null; // don't retain ref while waiting  
                    if (leader != null) // 4
                        available.await();
                    else {
                        Thread thisThread = Thread.currentThread();
                        leader = thisThread; // 5
                        try {
                            available.awaitNanos(delay); // 6
                        } finally {
                            if (leader == thisThread)
                                leader = null;
                        }
                    }
                }
            }
        } finally {
            if (leader == null && q.peek() != null) // 7  
                available.signal();
            lock.unlock();
        }
    }

    /**
     * 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();
        }
    }

 public int size() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return q.size();
        } finally {
            lock.unlock();
        }
    }