Java多线程15：Queue、BlockingQueue以及利用BlockingQueue实现生产者/消费者模型

红太狼 2022-05-25 07:39 88阅读 0赞

转自：http://www.cnblogs.com/xrq730/p/4855857.html

Queue是什么

队列，是一种数据结构。除了优先级队列和LIFO队列外，队列都是以FIFO（先进先出）的方式对各个元素进行排序的。无论使用哪种排序方式，队列的头都是调用remove()或poll()移除元素的。在FIFO队列中，所有新元素都插入队列的末尾。

Queue中的方法

Queue中的方法不难理解，6个，每2对是一个也就是总共3对。看一下JDK API就知道了：

![801753-20151005151001299-1024347662.png][]

注意一点就好，Queue通常不允许插入Null，尽管某些实现（比如LinkedList）是允许的，但是也不建议。

BlockingQueue

1、BlockingQueue概述

只讲BlockingQueue，因为BlockingQueue是Queue中的一个重点，并且通过BlockingQueue我们再次加深对于生产者/消费者模型的理解。其他的Queue都不难，通过查看JDK API和简单阅读源码完全可以理解他们的作用。

BlockingQueue，顾名思义，阻塞队列。BlockingQueue是在java.util.concurrent下的，因此不难理解，BlockingQueue是为了解决多线程中数据高效安全传输而提出的。

多线程中，很多场景都可以使用队列实现，比如经典的生产者/消费者模型，通过队列可以便利地实现两者之间数据的共享，定义一个生产者线程，定义一个消费者线程，通过队列共享数据就可以了。

当然现实不可能都是理想的，比如消费者消费速度比生产者生产的速度要快，那么消费者消费到 一定程度上的时候，必须要暂停等待一下了（使消费者线程处于WAITING状态）。BlockingQueue的提出，就是为了解决这个问题的，他不用程序员去控制这些细节，同时还要兼顾效率和线程安全。

阻塞队列所谓的"阻塞"，指的是某些情况下线程会挂起（即阻塞），一旦条件满足，被挂起的线程又会自动唤醒。使用BlockingQueue，不需要关心什么时候需要阻塞线程，什么时候需要唤醒线程，这些内容BlockingQueue都已经做好了

2、BlockingQueue中的方法

BlockingQueue既然是Queue的子接口，必然有Queue中的方法，上面已经列了。看一下BlockingQueue中特有的方法：

（1）void put(E e) throws InterruptedException

把e添加进BlockingQueue中，如果BlockingQueue中没有空间，则调用线程被阻塞，进入等待状态，直到BlockingQueue中有空间再继续

（2）void take() throws InterruptedException

取走BlockingQueue里面排在首位的对象，如果BlockingQueue为空，则调用线程被阻塞，进入等待状态，直到BlockingQueue有新的数据被加入

（3）int drainTo(Collection<? super E> c, int maxElements)

一次性取走BlockingQueue中的数据到c中，可以指定取的个数。通过该方法可以提升获取数据效率，不需要多次分批加锁或释放锁

3、ArrayBlockingQueue

基于数组的阻塞队列，必须指定队列大小。比较简单。ArrayBlockingQueue中只有一个ReentrantLock对象，这意味着生产者和消费者无法并行运行（见下面的代码）。另外，创建ArrayBlockingQueue时，可以指定ReentrantLock是否为公平锁，默认采用非公平锁。

/** Main lock guarding all access */
    private final ReentrantLock lock;
    /** Condition for waiting takes */
    private final Condition notEmpty;
    /** Condition for waiting puts */
    private final Condition notFull;

4、LinkedBlockingQueue

基于链表的阻塞队列，和ArrayBlockingQueue差不多。不过LinkedBlockingQueue如果不指定队列容量大小，会默认一个类似无限大小的容量，之所以说是类似是因为这个无限大小是Integer.MAX\_VALUE，这么说就好理解ArrayBlockingQueue为什么必须要制定大小了，如果ArrayBlockingQueue不指定大小的话就用Integer.MAX\_VALUE，那将造成大量的空间浪费，但是基于链表实现就不一样的，一个一个节点连起来而已。另外，LinkedBlockingQueue生产者和消费者都有自己的锁（见下面的代码），这意味着生产者和消费者可以"同时"运行。

![复制代码][copycode.gif]

/** Lock held by take, poll, etc */
    private final ReentrantLock takeLock = new ReentrantLock();
    
    /** Wait queue for waiting takes */
    private final Condition notEmpty = takeLock.newCondition();
    
    /** Lock held by put, offer, etc */
    private final ReentrantLock putLock = new ReentrantLock();
    
    /** Wait queue for waiting puts */
    private final Condition notFull = putLock.newCondition();

![复制代码][copycode.gif]

5、SynchronousQueue

比较特殊，一种没有缓冲的等待队列。什么叫做没有缓冲区，ArrayBlocking中有：

/** The queued items  */
    private final E[] items;

数组用以存储队列。LinkedBlockingQueue中有：

![复制代码][copycode.gif]

/**
     * Linked list node class
     */
    static class Node<E> {
        /** The item, volatile to ensure barrier separating write and read */
        volatile E item;
        Node<E> next;
        Node(E x) { item = x; }
    }

![复制代码][copycode.gif]

将队列以链表形式连接。

生产者/消费者操作数据实际上都是通过这两个"中介"来操作数据的，但是SynchronousQueue则是生产者直接把数据给消费者（消费者直接从生产者这里拿数据），好像又回到了没有生产者/消费者模型的老办法了。换句话说，每一个插入操作必须等待一个线程对应的移除操作。SynchronousQueue又有两种模式：

1、公平模式

采用公平锁，并配合一个FIFO队列（Queue）来管理多余的生产者和消费者

2、非公平模式

采用非公平锁，并配合一个LIFO栈（Stack）来管理多余的生产者和消费者，这也是SynchronousQueue默认的模式

利用BlockingQueue实现生产者消费者模型

上一篇我们写的生产者消费者模型有局限，局限体现在：

*  缓冲区内只能存放一个数据，实际生产者/消费者模型中的缓冲区内可以存放大量生产者生产出来的数据
 *  生产者和消费者处理数据的速度几乎一样

OK，我们就用BlockingQueue来简单写一个例子，并且让生产者、消费者处理数据速度不同。子类选择的是ArrayBlockingQueue，大小定为10：

![复制代码][copycode.gif]

public static void main(String[] args)
    {
        final BlockingQueue<String> bq = new ArrayBlockingQueue<String>(10);
        Runnable producerRunnable = new Runnable()
        {
            int i = 0;
            public void run()
            {
                while (true)
                {
                    try
                    {
                        System.out.println("我生产了一个" + i++);
                        bq.put(i + "");
                        Thread.sleep(1000);
                    } 
                    catch (InterruptedException e)
                    {
                        e.printStackTrace();
                    }
                }
            }
        };
        Runnable customerRunnable = new Runnable()
        {
            public void run()
            {
                while (true)
                {
                    try
                    {
                        System.out.println("我消费了一个" + bq.take());
                        Thread.sleep(3000);
                    } 
                    catch (InterruptedException e)
                    {
                        e.printStackTrace();
                    }
                }
            }
        };
        Thread producerThread = new Thread(producerRunnable);
        Thread customerThread = new Thread(customerRunnable);
        producerThread.start();
        customerThread.start();
    }

![复制代码][copycode.gif]

代码的做法是让生产者生产速度快于消费者消费速度的，看一下运行结果：

![复制代码][copycode.gif]

1 我生产了一个0
     2 我消费了一个1
     3 我生产了一个1
     4 我生产了一个2
     5 我消费了一个2
     6 我生产了一个3
     7 我生产了一个4
     8 我生产了一个5
     9 我消费了一个3
    10 我生产了一个6
    11 我生产了一个7
    12 我生产了一个8
    13 我消费了一个4
    14 我生产了一个9
    15 我生产了一个10
    16 我生产了一个11
    17 我消费了一个5
    18 我生产了一个12
    19 我生产了一个13
    20 我生产了一个14
    21 我消费了一个6
    22 我生产了一个15
    23 我生产了一个16
    24 我消费了一个7
    25 我生产了一个17
    26 我消费了一个8
    27 我生产了一个18

![复制代码][copycode.gif]

分两部分来看输出结果：

1、第1行~第23行。这块BlockingQueue未满，所以生产者随便生产，消费者随便消费，基本上都是生产3个消费1个，消费者消费速度慢

2、第24行~第27行，从前面我们可以看出，生产到16，消费到6，说明到了ArrayBlockingQueue的极限10了，这时候没办法，生产者生产一个ArrayBlockingQueue就满了，所以不能继续生产了，只有等到消费者消费完才可以继续生产。所以之后的打印内容一定是一个生产者、一个消费者

这就是前面一章开头说的"通过平衡生产者和消费者的处理能力来提高整体处理数据的速度"，这给例子应该体现得很明显。另外，也不要担心非单一生产者/消费者场景下的系统假死问题，缓冲区空、缓冲区满的场景BlockingQueue都是定义了不同的Condition，所以不会唤醒自己的同类。

[801753-20151005151001299-1024347662.png]: /images/20220525/b4daa352ee974d89876e31d23b589ac4.png
[copycode.gif]: /images/20220525/c845d9f79c244a3590140c5e8334f9ec.png