PriorityBlockingQueue和DelayQueue

阳光穿透心脏的1/2处 2022-09-03 05:19 128阅读 0赞

#### PriorityBlockingQueue ####

PriorityQueue（优先级阻塞队列）是按照元素的优先级从小到大出队列的。正因为如此， PriorityQueue元素之间需要可以比较大小（使用PriorityBlockingQueue的Comparator，或者元素实现Comparable）。

public class PriorityBlockingQueue<E> extends AbstractQueue<E> implements
    BlockingQueue<E>, java.io.Serializable {
        
    //...
    // 用数组实现的二插小根堆
    private transient Object[] queue; 
    //元素数量
    private transient int size;
    //定义了一个比较器
    private transient Comparator<? super E> comparator; 
    // 1个锁+一个条件，没有notfull条件，因为它会扩容
    private final ReentrantLock lock;
    private final Condition notEmpty;
    //...
    }

PriorityBlockingQueue如果不指定初始大小，内部会设定一个默认值11，当元素个数超过这个大小 之后，会自动扩容。

public PriorityBlockingQueue() {
        
       	this(DEFAULT_INITIAL_CAPACITY, null);
    }
    private static final int DEFAULT_INITIAL_CAPACITY = 11;
    
    public PriorityBlockingQueue(int initialCapacity,Comparator<? super E> comparator) {
        
    	if (initialCapacity < 1)
    	   	throw new IllegalArgumentException();
    	this.lock = new ReentrantLock();
    	this.notEmpty = lock.newCondition();
    	this.comparator = comparator;
    	this.queue = new Object[initialCapacity];
    }

put方法

public void put(E e) {
        
        offer(e); // never need to block
    }
    
    public boolean offer(E e) {
        
        if (e == null)
            throw new NullPointerException();
        final ReentrantLock lock = this.lock;
        lock.lock();
        int n, cap;
        Object[] array;
        //如果元素数量超过了数组的长度，就扩容
        while ((n = size) >= (cap = (array = queue).length))
        	//扩容
            tryGrow(array, cap);
        try {
        
            Comparator<? super E> cmp = comparator;
            //如果比较器为空使用元素的compareTo方法，比较并插入数组
            if (cmp == null)
                siftUpComparable(n, e, array);
            else
                siftUpUsingComparator(n, e, array, cmp);
            size = n + 1;
            //通知消费者消费 
            notEmpty.signal();
        } finally {
        
            lock.unlock();
        }
        return true;
    }

take方法

public E take() throws InterruptedException {
        
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        E result;
        try {
        
        	//出队列
            while ( (result = dequeue()) == null)
                notEmpty.await();
        } finally {
        
            lock.unlock();
        }
        return result;
    }
    
    private E dequeue() {
        
        int n = size - 1;
        if (n < 0)
            return null;
        else {
        
            Object[] array = queue;
            //这个数组构造了一个最小二叉堆，所以堆顶是要出队列的元素
            E result = (E) array[0];
            E x = (E) array[n];
            array[n] = null;
            Comparator<? super E> cmp = comparator;
            //出队列后重新调整二叉堆
            if (cmp == null)
                siftDownComparable(0, x, array, n);
            else
                siftDownUsingComparator(0, x, array, n, cmp);
            size = n;
            return result;
        }
    }

#### DelayQueue ####

DelayQueue（延迟队列）是一个按延迟时间从小到大出队的PriorityQueue。所谓延迟时间， 就是“未来将要执行的时间”减去“当前时间”。为此，放入DelayQueue中的元素，必须实现Delayed接口。

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
         */
         //如果getDelay的返回值小于或等于0，则说明该元素到期，需要从队列中拿出来执行
        long getDelay(TimeUnit unit);
    }

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 final Condition available = lock.newCondition();

take方法

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; // don't retain ref while waiting
                    //没到堆顶元素的延迟时间，不能直接返回
                    //如果有其他线程已经在等堆顶元素的延迟时间到点，则阻塞
                    if (leader != null)
                        available.await();
                    else {
        
                    	//如果没有其他线程等待，则自己是最先等待堆顶元素的延迟时间到点的线程
                        Thread thisThread = Thread.currentThread();
                        leader = thisThread;
                        try {
        
                        //阻塞delay的时间
                         available.awaitNanos(delay);
                        } finally {
        
                            if (leader == thisThread)
                                leader = null;
                        }
                    }
                }
            }
        } finally {
        
            if (leader == null && q.peek() != null)
            	//当前线程获取到了元素，唤醒其他线程
                available.signal();
            lock.unlock();
        }
    }

DelayQueue在调用take方法的时候不只是队列为空的时候才阻塞，如果堆顶元素的延迟时间没到，也会阻塞  
第一个等待堆顶元素的延迟时间的线程会阻塞等待一个有限的时间(available.awaitNanos(delay))，其他线程会阻塞等待signal

put方法的实现

public void put(E e) {
        
        offer(e);
    }
    public boolean offer(E e) {
        
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
        
        	//元素放进二叉堆
            q.offer(e);
            //如果新元素在堆顶，说明延迟最小
            if (q.peek() == e) {
        
            	//把之前take方法等待堆顶延迟时间的线程取消
                leader = null;
                //通知等待的线程消费
                available.signal();
            }
            return true;
        } finally {
        
            lock.unlock();
        }
    }

put的时候，只有新元素是堆顶的元素（延迟最小的元素），才通知消费线程消费（available.signal()），如果新元素不是堆顶的元素，那么就不用通知消费线程消费，因为take方法中，已经有available.awaitNanos(delay)的线程在等待消费堆顶的元素了。