【原创】Java并发编程系列31 | 阻塞队列（上）

阻塞队列在并发编程非常常用，被广泛使用在“生产者-消费者”问题中。接下来两篇文章就来详细介绍阻塞队列。本文是阻塞队列上篇。

介绍
基本操作
应用
常用阻塞队列及源码 4.1 ArrayBlockingQueue 4.2 LinkedBlockingQueue 4.3 SynchronousQueue 4.4 PriorityBlockingQueue 4.5 DelayQueue

1. 介绍

阻塞队列（BlockingQueue）是一个比普通队列多出两个附加操作的队列。两个操作分别是：

在队列为空时，获取元素的线程会等待队列变为非空。
当队列满时，存储元素的线程会等待队列可用。

阻塞队列（BlockingQueue）被广泛使用在“生产者-消费者”问题中。当队列容器已满，生产者线程会被阻塞，直到队列未满；当队列容器为空时，消费者线程会被阻塞，直至队列非空时为止。这样可以对各个模块的业务功能进行解耦，生产者将“生产”出来的数据放置在数据容器中，而消费者仅仅只需要在“数据容器”中进行获取数据即可，这样生产者线程和消费者线程就能够进行解耦，只专注于自己的业务功能即可。

2. 基本操作

BlockingQueue基本操作如下：

// 插入元素
add(E e) ：往队列插入数据，当队列满时，插入元素时会抛出IllegalStateException异常；
offer(E e)：当往队列插入数据时，插入成功返回true，否则则返回false。当队列满时不会抛出异常；

// 删除元素
remove(Object o)：从队列中删除数据，成功则返回true，否则为false
poll：删除数据，当队列为空时，返回null；

// 查看元素
element：获取队头元素，如果队列为空时则抛出NoSuchElementException异常；
peek：获取队头元素，如果队列为空则抛出NoSuchElementException异常

// 插入数据：
put()：当阻塞队列容量已经满时，往阻塞队列插入数据的线程会被阻塞，直至阻塞队列已经有空余的容量可供使用；
offer(E e, long timeout, TimeUnit unit)：若阻塞队列已经满时，同样会阻塞插入数据的线程，直至阻塞队列已经有空余的地方，与put方法不同的是，该方法会有一个超时时间，若超过当前给定的超时时间，插入数据的线程会退出；

// 删除数据：
take()：当阻塞队列为空时，获取队头数据的线程会被阻塞；
poll(long timeout, TimeUnit unit)：当阻塞队列为空时，获取数据的线程会被阻塞，另外，如果被阻塞的线程超过了给定的时长，该线程会退出

put(e) 和 take() 是BlockingQueue的核心方法，也是我们比较关注的。

3. 应用

使用普通队列实现生产者-消费者模式，代码如下：

public class BlockingDemo {
    static LinkedList<Integer> queue = new LinkedList<Integer>();
    static int maxSize = 5;

    public static void main(String[] args) throws Exception {
        new Thread("生产者") {
            public void run() {
                while (true) {
                    synchronized (queue) {
                        while (queue.size() >= maxSize) {
                            try {
                                System.out.println("队列满了。。。");
                                queue.wait();
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        }
                        queue.addLast(1);
                        queue.notify();
                        System.out.println("队列中添加了一个元素, size=" + queue.size());
                    }
                }
            };
        }.start();

        new Thread("消费者") {
            public void run() {
                while (true) {
                    synchronized (queue) {
                        while (queue.size() <= 0) {
                            try {
                                System.out.println("队列空了。。。");
                                queue.wait();
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        }
                        queue.removeFirst();
                        queue.notify();
                        System.out.println("队列中删除了一个元素, size=" + queue.size());
                    }
                }
            };
        }.start();
    }
}

控制台输出：

队列中添加了一个元素, size=1
队列中添加了一个元素, size=2
队列中添加了一个元素, size=3
队列中添加了一个元素, size=4
队列中添加了一个元素, size=5
队列中删除了一个元素, size=4
队列中删除了一个元素, size=3
队列中删除了一个元素, size=2
队列中删除了一个元素, size=1
队列中删除了一个元素, size=0
队列空了。。。
队列中添加了一个元素, size=1
队列中添加了一个元素, size=2
队列中添加了一个元素, size=3
队列中添加了一个元素, size=4
队列中添加了一个元素, size=5
队列满了。。。
队列中删除了一个元素, size=4
队列中删除了一个元素, size=3
队列中删除了一个元素, size=2
队列中删除了一个元素, size=1
队列中删除了一个元素, size=0
队列空了。。。
队列中添加了一个元素, size=1
队列中添加了一个元素, size=2
队列中添加了一个元素, size=3
队列中添加了一个元素, size=4
队列中添加了一个元素, size=5
队列满了。。。
...部分省略...

使用阻塞队列实现生产者消费者模式：

public class Test {
    static ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<Integer>(5);

    public static void main(String[] args) throws Exception {
        new Thread("生产者") {
            public void run() {
                while (true) {
                    try {
                        queue.put(1);
                        System.out.println("队列中添加了一个元素, size=" + queue.size());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
        }.start();

        Thread.sleep(500);
        
        new Thread("消费者") {
            public void run() {
                while (true) {
                    try {
                        queue.take();
                        System.out.println("队列中删除了一个元素, size=" + queue.size());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
        }.start();
    }
}

输出结果如下：

队列中添加了一个元素, size=1
队列中添加了一个元素, size=2
队列中添加了一个元素, size=3
队列中添加了一个元素, size=4
队列中添加了一个元素, size=5
队列中删除了一个元素, size=4
队列中添加了一个元素, size=5
队列中删除了一个元素, size=4
队列中添加了一个元素, size=5
队列中删除了一个元素, size=4
队列中添加了一个元素, size=5
队列中删除了一个元素, size=4
队列中添加了一个元素, size=5
队列中添加了一个元素, size=5
队列中删除了一个元素, size=4
队列中删除了一个元素, size=4
队列中添加了一个元素, size=5
队列中删除了一个元素, size=4
队列中添加了一个元素, size=5
队列中删除了一个元素, size=4
...部分省略...

4. 常用阻塞队列

4.1 ArrayBlockingQueue

ArrayBlockingQueue是由数组实现的有界队列，通过ReentrantLock锁保证队列数据的安全性，通过ReentrantLock的条件Condition是实现阻塞。
添加元素时，如果队列满了不能添加元素，就将添加元素的线程阻塞并加入notFull条件队列；当成功删除元素后，队列就可以添加元素了，唤醒notFull条件队列中阻塞的线程，添加元素。
删除元素时，如果队列空了不能删除元素，就将删除元素的线程阻塞并加入notEmpty条件队列；当成功添加元素后，队列就可以删除元素了，唤醒notEmpty条件队列中阻塞的线程，删除元素。

类结构

ArrayBlockingQueue是由数组实现的有界队列，通过ReentrantLock锁保证队列数据的安全性，通过ReentrantLock的条件Condition是实现阻塞。

队列创建时，确定队列大小和是否公平。

源码：

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
    implements BlockingQueue<E>, java.io.Serializable {
    final Object[] items;// 用于存放元素的数组
    int takeIndex;// 下一次读取操作的位置
    int putIndex;// 下一次写入操作的位置
    int count;// 队列中的元素数量
    
    // 通过lock及其两个条件notEmpty、notFull控制阻塞
    final ReentrantLock lock;
    private final Condition notEmpty;
    private final Condition notFull;
    
    // 创建队列时，确定队列大小和是否公平
    public ArrayBlockingQueue(int capacity, boolean fair) {
 if (capacity <= 0)
  throw new IllegalArgumentException();
 this.items = new Object[capacity];
 lock = new ReentrantLock(fair);
 notEmpty = lock.newCondition();
 notFull =  lock.newCondition();
    }
}

put()

获取锁lock
队列满时，将当前线程加入notFull条件队列阻塞；当有元素出队时，队列就不满了，可以让元素入队了，此时会唤醒notFull条件队列中的线程，加入AQS阻塞队列等锁或者直接抢锁，然后将元素入队。
入队：入队成功之后，唤醒notEmpty条件队列中阻塞的线程，让其元素出队
释放锁lock

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
    implements BlockingQueue<E>, java.io.Serializable {
    final Object[] items;// 用于存放元素的数组
    int takeIndex;// 下一次读取操作的位置
    int putIndex;// 下一次写入操作的位置
    int count;// 队列中的元素数量
    
    // 通过lock及其两个条件notEmpty、notFull控制阻塞
    final ReentrantLock lock;
    private final Condition notEmpty;
    private final Condition notFull;
}


public void put(E e) throws InterruptedException {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();// 获取lock锁
    try {
        /*
         * 队列满时，将当前线程加入notFull条件队列阻塞；
         * 当有元素出队时，队列就不满了，可以让元素入队了，
         * 此时会唤醒notFull条件队列中的线程，加入AQS阻塞队列等锁或者直接抢锁，然后将元素入队。
         */
        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条件队列中阻塞的线程，让其元素出队
    notEmpty.signal();
}

take()

获取锁lock
队列空时，将当前线程加入notEmpty条件队列阻塞；当有元素入队时，队列不为空了就可以take出元素，此时会唤醒notEmpty条件队列中的线程，加入AQS阻塞队列等锁或者直接抢锁，然后执行出队操作。
出队：出队成功，唤醒notFull条件队列中阻塞的线程，让其元素入队
释放锁lock

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();// 获取lock锁
    try {
        /*
         * 队列空时，将当前线程加入notEmpty条件队列阻塞；
         * 当有元素入队时，队列不为空了就可以take出元素，
         * 此时会唤醒notEmpty条件队列中的线程，加入AQS阻塞队列等锁或者直接抢锁，然后执行出队操作。
         */
        while (count == 0)
            notEmpty.await();
        return dequeue();// 出队
    } finally {
        lock.unlock();// 解锁
    }
}

private E dequeue() {
    final Object[] items = this.items;
    @SuppressWarnings("unchecked")
    E x = (E) items[takeIndex];
    items[takeIndex] = null;
    if (++takeIndex == items.length)
        takeIndex = 0;
    count--;
    if (itrs != null)
        itrs.elementDequeued();
    // 出队成功，唤醒notFull条件队列中阻塞的线程，让其元素入队
    notFull.signal();
    return x;
}

4.2 LinkedBlockingQueue

LinkedBlockingQueue用链表实现的有界阻塞队列。（不设置容量，默认为Integer.MAX_VALUE）

锁takeLock保证删除数据的安全性，队列为空时读操作线程阻塞并加入takeLock锁的notEmpty条件等待队列。
锁putLock保证添加数据的安全性，队列满时写操作线程阻塞并加入putLock锁的notFull条件等待队列。

ArrayBlockingQueue的读写使用同一个锁来保证数据安全。LinkedBlockingQueue的读写分别用不同的锁来保证数据安全，采用不同的锁可以使读线程和写线程并发执行，提高了吞吐量，但也增加了编程的复杂度。

类结构

LinkedBlockingQueue用链表实现的有界阻塞队列。（不设置容量，默认为Integer.MAX_VALUE）

锁takeLock保证删除数据的安全性，队列为空时读操作线程阻塞并加入takeLock锁的notEmpty条件等待队列。

锁putLock保证添加数据的安全性，队列满时写操作线程阻塞并加入putLock锁的notFull条件等待队列。

// 节点类 单向链表
static class Node<E> {
 E item;
 Node<E> next;
 Node(E x) { item = x; }
}

private final int capacity;// 队列容量 不设置默认为Integer.MAX_VALUE
private final AtomicInteger count = new AtomicInteger(0);// 队列中的元素数量
private transient Node<E> head;// 队头
private transient Node<E> last;// 队尾

// take, poll, peek 等读操作的方法需要获取到这个锁
private final ReentrantLock takeLock = new ReentrantLock();
// 如果读操作的时候队列是空的，加入notEmpty等待队列
private final Condition notEmpty = takeLock.newCondition();

// put, offer 等写操作的方法需要获取到这个锁
private final ReentrantLock putLock = new ReentrantLock();
// 如果写操作的时候队列是满的，加入notFull等待队列
private final Condition notFull = putLock.newCondition();

// 有界队列， 不设置容量，默认为Integer.MAX_VALUE
public LinkedBlockingQueue() {
 this(Integer.MAX_VALUE);
}

public LinkedBlockingQueue(int capacity) {
 if (capacity <= 0) throw new IllegalArgumentException();
 this.capacity = capacity;
 last = head = new Node<E>(null);
}

put(E e)

获取putLock锁
如果队列满，当前线程阻塞并加入notFull条件等待队列
入队
这个元素入队成功后，队列还没有满，唤醒notFull队列中等待添加元素的线程。（因为添加元素和删除元素不是用的同一个锁导致会有这种情况发生）
释放掉putLock锁
如果添加元素前队列为空，可能会有读线程阻塞，所以在这个元素入队后，就唤醒阻塞的读线程

public void put(E e) throws InterruptedException {
    if (e == null) throw new NullPointerException();
    int c = -1;
    Node<E> node = new Node(e);
    final ReentrantLock putLock = this.putLock;
    final AtomicInteger count = this.count;
    putLock.lockInterruptibly();// 获取putLock锁
    try {
        // 如果队列满，当前线程阻塞并加入notFull条件等待队列
        while (count.get() == capacity) {
            notFull.await();
        }
        enqueue(node);// 入队
        c = count.getAndIncrement();// count 原子加 1，注意：这里返回的c是原来的值，并不是加1后的值。
        /*
         * 这个元素入队成功后，队列还没有满，notFull.signal() 唤醒notFull队列中等待添加元素的线程。
         * 为什么队列还没有满，但是添加元素线程却在阻塞状态呢？
         * 因为添加元素和删除元素不是用的同一个锁，所以添加元素和删除元素是可以同时进行的。
         * 当添加元素时发现队列满了，线程阻塞。此时另一个线程执行删除操作，队列又不满了。
         * 于是出现了这个情况：队列还没有满，但是添加元素线程却在阻塞状态。
         */
        if (c + 1 < capacity)
            notFull.signal();
    } finally {
        putLock.unlock();// 入队后，释放掉 putLock
    }
    /*
     * c == 0表示队列在这个元素入队前是空的，队列为空时可能会有读线程阻塞
     * 所以在这个元素入队后，就唤醒阻塞的读线程
     */
    if (c == 0)
        signalNotEmpty();
}

/**
 * 队列尾部插入元素
 */
private void enqueue(Node<E> node) {
    last = last.next = node;
}

/**
 * 唤醒读线程
 */
private void signalNotEmpty() {
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lock();
    try {
        notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
}

take()

获取takeLock锁
如果队列空，当前线程阻塞并加入notEmpty条件等待队列
出队
这个元素出队成功后，队列还有元素，唤醒notEmpty队列中等待删除元素的线程。（因为添加元素和删除元素不是用的同一个锁导致会有这种情况发生）
释放takeLock锁
如果添加元素前队列是满的，可能有写线程阻塞等待，所以唤醒写线程

public E take() throws InterruptedException {
    E x;
    int c = -1;
    final AtomicInteger count = this.count;
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lockInterruptibly();// 获取锁takeLock
    try {
        // 如果队列为空，当前线程阻塞并加入notEmpty条件等待队列
        while (count.get() == 0) {
            notEmpty.await();
        }
        x = dequeue();// 出队
        c = count.getAndDecrement();// count 进行原子减 1,注意：这里返回的c是原来的值，并不是减1后的值。
        /*
         * 这个元素出队成功后，队列还没有满，notEmpty.signal() 唤醒notEmpty队列中等待删除元素的线程。
         * 当队列中还有元素时，为什么会有读线程在阻塞呢？
         * 因为添加元素和删除元素不是用的同一个锁，所以添加元素和删除元素是可以同时进行的。
         * 当删除元素时发现队列空了，线程阻塞。此时另一个线程执行添加操作，队列又不空了。
         * 于是出现了这个情况：当队列中还有元素时，会有读线程在阻塞状态。
         */
        if (c > 1)
            notEmpty.signal();
    } finally {
        takeLock.unlock();// 释放锁takeLock
    }
    // c == capacity表示删除元素之前队列是满的，队列满时可能有写线程阻塞等待，所以唤醒写线程
    if (c == capacity)
        signalNotFull();
    return x;
}
// 出队
private E dequeue() {
    Node<E> h = head;
    Node<E> first = h.next;
    h.next = h; // help GC
    head = first;
    E x = first.item;
    first.item = null;
    return x;
}

// 唤醒写线程来写
private void signalNotFull() {
    final ReentrantLock putLock = this.putLock;
    putLock.lock();
    try {
        notFull.signal();
    } finally {
        putLock.unlock();
    }
}