BlockingQueue

r囧r小猫 2021-11-23 11:06 349阅读 0赞

ArrayBlockingQueue：一锁，两Condition，takeIndex，putIndex，count

public E take() throws InterruptedException {
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
                while (count == 0)  // 若被唤醒并获取锁后依然需要判断，故循环
                    notEmpty.await();
                return dequeue(); 
            } finally {
                lock.unlock();
            }
        }
     private E dequeue() {
            final Object[] items = this.items;
            E x = (E) items[takeIndex];
            items[takeIndex] = null;
            if (++takeIndex == items.length) //Index 最大为length-1，否则跳回首部
                takeIndex = 0;
            count--;
            notFull.signal();
            return x;
        }
    public void put(E e) throws InterruptedException {
            checkNotNull(e);
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
                while (count == items.length)
                    notFull.await();
                enqueue(e);
            } finally {
                lock.unlock();
            }
        }
        
         private void enqueue(E x) {
            final Object[] items = this.items;
            items[putIndex] = x;
            if (++putIndex == items.length)
                putIndex = 0;
            count++;
            notEmpty.signal();
        }

LinkedBlockingQueue：单向链表，含首尾节点引用，双锁双Condition(takeLock-NotEmpty, putLock-NotFull), count, capacity  
双锁提升性能，避免put和take在单锁上竞争

public void put(E e) throws InterruptedException {
            int c = -1;
            Node<E> node = new Node<E>(e);
            final ReentrantLock putLock = this.putLock;
            final AtomicInteger count = this.count;
            putLock.lockInterruptibly();
            try {
                while (count.get() == capacity) {
                    notFull.await();
                }
                enqueue(node);
                c = count.getAndIncrement();
                if (c + 1 < capacity)
                    notFull.signal(); // 若发现不为满，可继续唤醒put线程
            } finally {
                putLock.unlock();
            }
            if (c == 0)
                signalNotEmpty(); // 取takeLock，notEmpty.signal()，只在原来为空的情况下去唤醒，避免不必要唤醒
        }
    
     public E take() throws InterruptedException {
            E x;
            int c = -1;
            final AtomicInteger count = this.count;
            final ReentrantLock takeLock = this.takeLock;
            takeLock.lockInterruptibly();
            try {
                while (count.get() == 0) {
                    notEmpty.await();
                }
                x = dequeue();
                c = count.getAndDecrement();
                if (c > 1)
                    notEmpty.signal();
            } finally {
                takeLock.unlock();
            }
            if (c == capacity)
                signalNotFull();
            return x;

DelayQueue：无界阻塞队列，单锁单Condition，内置PriorityQueue，leader变量记录当前是否有线程在队列上等待，存放Delayed元素（继承Comparable），元素按自然排序（Comparable）放入PriorityQueue（取出的是小的，如1，2取出1），从PriorityQueue取出且getDelay小于等于0才能真正取出

TransferQueue(扩展了BlockingQueue)：实现LinkedTransferQueue，单向链表，其内部Node可能是consumer(带线程）或producer(带数据)，可以让最早开始等待线程最先获取数据  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2tvYmV4emY_size_16_color_FFFFFF_t_70]

PriorityBlockingQueue：数组（自动扩容），单锁单Condition(NotEmpty，由于扩容所有没有NotFull)，数组实现了堆结构  
取出：取堆顶元素并向下调整，放入：放入堆尾并向上调整

SynchronousQueue：适用于直接交付，而不是保存元素，公平模式使用TransferQueue，非公平TransferStack，这里的TransferQueue和上面的不同，put线程（带数据，带线程）无法立即交付则阻塞，get线程（带线程）无法立即交付则阻塞

PriorityBlockingQueue，PriorityQueue，DelayQueue，ScheduledThreadPoolExecutor内的DelayedWorkQueue都是无界的（数组扩容），DelayQueue和DelayedWorkQueue极其相似

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2tvYmV4emY_size_16_color_FFFFFF_t_70]: /images/20211122/fc770d077b804e419a70b396c4d1a767.png