阻塞队列

[cnwutianhao]

介绍阻塞队列的定义、种类、实现原理以及应用。

一、阻塞队列简介

阻塞队列常用于生产者和消费者的场景，生产者是往队列里添加元素的线程，消费者是从队列里拿元素的线程。阻塞队列就是生产者存放元素的容器，而消费者也只从容器里拿元素。

1. 常见阻塞场景

   阻塞队列有两个常见的阻塞场景，它们分别是：
   
   （1）当队列中没有数据的情况下，消费者端的所有线程都会被自动阻塞（挂起），直到有数据放入队列。
   
   （2）当队列中填满数据的情况下，生产者端的所有线程都会被自动阻塞（挂起），直到队列中有空的位置，线程被自动唤醒。

   支持以上两种阻塞场景的队列被称为阻塞队列。

2. BlockingQueue 的核心方法

   放入数据：

   • offer(anObject)：表示如果可能的话，将 anObject 加到 BlockingQueue 里。即如果 BlockingQueue 可以容纳，则返回 true，否则返回 false。（本方法不阻塞当前执行方法的线程。）
   • offer(E o, long timeout, TimeUnit unit)：可以设定等待的时间。如果在指定的时间内还不能往队列中加入 BlockingQueue，则返回失败。
   • put(anObject)：将 anObject 加到 BlockingQueue 里。如果 BlockQueue 没有空间，则调用此方法的线程被阻断，直到 BlockingQueue 里面有空间再继续。

   获取数据：

   • poll(time)：取走 BlockingQueue 里排在首位的对象。若不能立即取出，则可以等 time 参数规定的时间。取不到时返回 null。
   • poll(long timeout, TimeUnit unit)：从 BlockingQueue 中取出一个队首的对象。如果在指定时间内，队列一旦有数据可取，则立即返回队列中的数据；否则直到时间超时还没有数据可取，返回失败。
   • take()：取走 BlockingQueue 里排在首位的对象。若 BlockingQueue 为空，则阻断进入等待状态，直到 BlockingQueue 有新的数据被加入。
   • drainTo()：一次性从 BlockingQueue 获取所有可用的数据对象（还可以指定获取数据的个数）。通过该方法，可以提升获取数据的效率；无须多次分批加锁或释放锁。

二、Java 中的阻塞队列

在 Java 中提供了7个阻塞队列，它们分别如下所示：

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

下面分别介绍这些阻塞队列：

1. ArrayBlockingQueue

   它是用数组实现的有界阻塞队列，并按照先进先出（FIFO）的原则对元素进行排序。默认情况下不保证线程公平地访问队列。公平访问队列就是指阻塞的所有生产者线程或消费者线程，当队列可用时，可以按照阻塞的先后顺序访问队列。即先阻塞的生产者线程，可以先往队列里插入元素；先阻塞的消费者线程，可以先从队列里获取元素。通常情况下为了保证公平性会降低吞吐量。我们可以使用以下代码创建一个公平的阻塞队列，如下所示：

   ArrayBlockingQueue fairQueue = new ArrayBlockingQueue(2000, true);

2. LinkedBlockingQueue

   它是基于链表的阻塞队列，同 ArrayListBlockingQueue 类似，此队列按照先进先出（FIFO）的原则对元素进行排序，其内部也维持着一个数据缓冲队列（该队列由一个链表构成）。当生产者往队列中放入一个数据时，队列会从生产者手中获取数据，并缓存在队列内部，而生产者立即返回；只有当队列缓冲区达到缓存容量的最大值时（LinkedBlockingQueue 可以通过构造方法指定该值），才会阻塞生产者队列，直到消费者从队列中消费掉一份数据，生产者线程会被唤醒。反之，对于消费者这端的处理也基于同样的原理。而 LinkedBlockingQueue 之所以能够高效地处理并发数据，还因为其对于生产者端和消费者端分别采用了独立的锁来控制数据同步。这也意味着在高并发的情况下生产者和消费者可以并行地操作队列中的数据，以此来提高整个队列的并发性能。作为开发者，我们需要注意的是，如果构造一个 LinkedBlockingQueue 对象，而没有指定其容量大小，LinkedBlockingQueue 会默认一个类似无限大小的容量（Integer.MAX_VALUE）。这样一来，如果生产者的速度一旦大于消费者的速度，也许还没有等到队列满阻塞产生，系统内存就有可能已被消耗殆尽了。ArrayBlockingQueue 和 LinkedBlockingQueue 是两个最普通也是最常用的阻塞队列。一般情况下，在处理多线程间的生产者-消费者问题时，使用这两个类足已。

3. PriorityBlockingQueue

   它是一个支持优先级的无界队列。默认情况下元素采取自然顺序升序排列。这里可以自定义实现 compareTo() 方法来指定元素进行排序规则；或者初始化 PriorityBlockingQueue 时，指定构造参数 Comparator 来对元素进行排序。但其不能保证同优先级元素的顺序。

4. DelayQueue

   它是一个支持延时获取元素的无界阻塞队列。队列使用 PriorityQueue 来实现。队列中的元素必须实现 Delayed 接口。创建元素时，可以指定元素到期的时间，只有在元素到期时才能从队列中取走。

5. SynchronousQueue

   它是一个不存储元素的阻塞队列。每个插入操作必须等待另一个线程的移除操作，同样任何一个移除操作都等待另一个线程的插入操作。因此此队列内部其实没有任何一个元素，或者说容量是0，严格来说它并不是一种容器。由于队列没有容量，因此不能调用 peek 操作（返回队列的头元素）。

6. LinkedTransferQueue

   它是一个由链表结构组成的无界阻塞 TransferQueue 队列。LinkedTransferQueue 实现了一个重要的接口 TransferQueue。该接口含有5个方法，其中有3个重要的方法，它们分别如下所示：
   
   （1）transfer(E e)：若当前存在一个正在等待获取的消费者线程，则立刻将元素传递给消费者；如果没有消费者在等待接收数据，就会将元素插入到队列尾部，并且等待进入阻塞状态，直到有消费者线程取走该元素。
   
   （2）tryTransfer(E e)：若当前存在一个正在等待获取的消费者线程，则立刻将元素传递给消费者；若不存在，则返回 false，并且不进入队列，这是一个不阻塞的操作。与 transfer 方法不同的是，tryTransfer 方法无论消费者是否接收，其都会立即返回；而 transfer 方法则是消费者接收了才返回。
   
   （3）tryTransfer(E e, long timeout, TimeUnit unit)：若当前存在一个正在等待获取的消费者线程，则立刻将元素传递给消费者；若不存在则将元素插入到队列尾部，并且等待消费者线程取走该元素。若在指定的超时时间内元素未被消费者线程获取，则返回 false；若在指定的超时时间内其被消费者线程获取，则返回 true。

7. LinkedBlockingDeque

   它是一个由链表结构组成的双向阻塞队列。双向队列可以从队列的两端插入和移出元素，因此在多线程同时入队时，也就减少了一半的竞争。由于是双向的，因此 LinkedBlockingDeque 多了 addFirst、addLast、offerFirst、offerLast、peekFirst、peekLast 等方法。其中，以 First 单词结尾的方法，表示插入、获取或移除双端队列的第一个元素；以 Last 单词结尾的方法，表示插入、获取或移除双端队列的最后一个元素。

三、阻塞队列的实现原理

以 ArrayBlockingQueue 为例，我们先来看看源码，如下所示：

public class ArrayBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable {

    private static final long serialVersionUID = -817911632652898426L;
    final Object[] items;
    int takeIndex;
    int putIndex;
    int count;
    final ReentrantLock lock;
    private final Condition notEmpty;
    private final Condition notFull;

    ...
}

从上面的代码可以看出 ArrayBlockingQueue 是维护一个 Object 类型的数组，takeIndex 和 putIndex 分别表示队首元素和队尾元素的下标，count 表示队列中元素的个数，lock 则是一个可重入锁，notEmpty 和 notFull 是等待条件。

接下来我们看看关键方法 put，源码如下所示：

public void put(E e) throws InterruptedException {
    Objects.requireNonNull(e);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == items.length)
            notFull.await();
        enqueue(e);
    } finally {
        lock.unlock();
    }
}

从 put 方法的实现可以看出，它先获取了锁，并且获取的是可中断锁，然后判断当前元素个数是否等于数组的长度，如果相等，则调用 notFull.await() 进行等待。当此线程被其他线程唤醒时，通过 enqueue(e) 方法插入元素，最后解锁。

接下来看看 enqueue(e) 方法，如下所示：

private void enqueue(E e) {
    final Object[] items = this.items;
    items[putIndex] = e;
    if (++putIndex == items.length) putIndex = 0;
    count++;
    notEmpty.signal();
}

插入成功后，通过 notEmpty 唤醒正在等待取元素的线程。

再来看看 take 方法，如下所示：

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0)
            notEmpty.await();
        return dequeue();
    } finally {
        lock.unlock();
    }
}

跟 put 方法实现类似，put 方法等待的是 notFull 信号，而 take 方法等待的是 notEmpty 信号。在 take 方法中，如果可以取元素，则通过 dequeue 方法取得元素。

下面是 dequeue 方法的实现：

private E dequeue() {
    final Object[] items = this.items;
    @SuppressWarnings("unchecked")
    E e = (E) items[takeIndex];
    items[takeIndex] = null;
    if (++takeIndex == items.length) takeIndex = 0;
    count--;
    if (itrs != null)
        itrs.elementDequeued();
    notFull.signal();
    return e;
}

跟 enqueue 方法类似，在获取元素后，通过 notFull 的 signal 方法来唤醒正在等待插入元素的线程。

四、阻塞队列的使用场景

除了线程池的实现使用阻塞队列外，我们还可以在生产者-消费者模式中使用阻塞队列：首先使用 Object.wait()、Object.notify() 和非阻塞队列实现生产者-消费者模式，代码如下所示：

public class Test {
    private int queueSize = 10;

    private PriorityQueue<Integer> queue = new PriorityQueue<>(queueSize);

    public static void main(String[] args) {
        Test test = new Test();
        Producer producer = test.new Producer();
        Consumer consumer = test.new Consumer();
        producer.start();
        consumer.start();
    }

    class Consumer extends Thread {
        @Override
        public void run() {
            while (true) {
                synchronized (queue) {
                    while (queue.isEmpty()) {
                        try {
                            System.out.println("队列空，等待数据");
                            queue.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                            queue.notify();
                        }
                    }

                    // 每次移走队首元素
                    queue.poll();
                    queue.notify();
                }
            }
        }
    }

    class Producer extends Thread {
        @Override
        public void run() {
            while (true) {
                synchronized (queue) {
                    while (queue.size() == queueSize) {
                        try {
                            System.out.println("队列满，等待有空余空间");
                            queue.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                            queue.notify();
                        }
                    }

                    // 每次插入一个元素
                    queue.offer(1);
                    queue.notify();
                }
            }
        }
    }
}

下面是使用阻塞队列实现的生产者-消费者模式，代码如下所示：

public class Test {
    private int queueSize = 10;

    private ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<>(queueSize);

    public static void main(String[] args) {
        Test test = new Test();
        Producer producer = test.new Producer();
        Consumer consumer = test.new Consumer();
        producer.start();
        consumer.start();
    }

    class Consumer extends Thread {
        @Override
        public void run() {
            while (true) {
                try {
                    queue.take();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    class Producer extends Thread {
        @Override
        public void run() {
            while (true) {
                try {
                    queue.put(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

很显然，使用阻塞队列实现无须单独考虑同步和线程间通信的问题，其实现起来很简单。