生产者消费者问题

简介

生产者消费者模式并不是GOF提出的23种设计模式之一，23种设计模式都是建立在面向对象的基础之上的，但其实面向过程的编程中也有很多高效的编程模式，生产者消费者模式便是其中之一，它是我们编程过程中最常用的一种设计模式。

在实际的软件开发过程中，经常会碰到如下场景：某个模块负责产生数据，这些数据由另一个模块来负责处理（此处的模块是广义的，可以是类、函数、线程、进程等）。产生数据的模块，就形象地称为生产者；而处理数据的模块，就称为消费者。

单单抽象出生产者和消费者，还够不上是生产者／消费者模式。该模式还需要有一个缓冲区处于生产者和消费者之间，作为一个中介。生产者把数据放入缓冲区，而消费者从缓冲区取出数据

为了不至于太抽象，我们举一个寄信的例子（虽说这年头寄信已经不时兴，但这个例子还是比较贴切的）。假设你要寄一封平信，大致过程如下：

    1、你把信写好——相当于生产者制造数据

    2、你把信放入邮筒——相当于生产者把数据放入缓冲区

    3、邮递员把信从邮筒取出——相当于消费者把数据取出缓冲区

    4、邮递员把信拿去邮局做相应的处理——相当于消费者处理数据

缓存区

缓存区的作用

 解 耦

假设生产者和消费者分别是两个类。如果让生产者直接调用消费者的某个方法，那么生产者对于消费者就会产生依赖（也就是耦合）。将来如果消费者的代码发生变化，可能会影响到生产者。而如果两者都依赖于某个缓冲区，两者之间不直接依赖，耦合也就相应降低了。

接着上述的例子，如果不使用邮筒（也就是缓冲区），你必须得把信直接交给邮递员。有同学会说，直接给邮递员不是挺简单的嘛？其实不简单，你必须得认识谁是邮递员，才能把信给他（光凭身上穿的制服，万一有人假冒，就惨了）。这就产生和你和邮递员之间的依赖（相当于生产者和消费者的强耦合）。万一哪天邮递员换人了，你还要重新认识一下（相当于消费者变化导致修改生产者代码）。而邮筒相对来说比较固定，你依赖它的成本就比较低（相当于和缓冲区之间的弱耦合）。

 支持并发（concurrency）

生产者直接调用消费者的某个方法，还有另一个弊端。由于函数调用是同步的（或者叫阻塞的），在消费者的方法没有返回之前，生产者只好一直等在那边。万一消费者处理数据很慢，生产者就会白白糟蹋大好时光。

使用了生产者／消费者模式之后，生产者和消费者可以是两个独立的并发主体（常见并发类型有进程和线程两种，后面的帖子会讲两种并发类型下的应用）。生产者把制造出来的数据往缓冲区一丢，就可以再去生产下一个数据。基本上不用依赖消费者的处理速度。

其实当初这个模式，主要就是用来处理并发问题的。

从寄信的例子来看。如果没有邮筒，你得拿着信傻站在路口等邮递员过来收（相当于生产者阻塞）；又或者邮递员得挨家挨户问，谁要寄信（相当于消费者轮询）。不管是哪种方法，都挺土的。

支持忙闲不均

缓冲区还有另一个好处。如果制造数据的速度时快时慢，缓冲区的好处就体现出来了。当数据制造快的时候，消费者来不及处理，未处理的数据可以暂时存在缓冲区中。等生产者的制造速度慢下来，消费者再慢慢处理掉。

为了充分复用，我们再拿寄信的例子来说事。假设邮递员一次只能带走1000封信。万一某次碰上情人节（也可能是圣诞节）送贺卡，需要寄出去的信超过1000封，这时候邮筒这个缓冲区就派上用场了。邮递员把来不及带走的信暂存在邮筒中，等下次过来时再拿走。

何谓数据单元捏？简单地说，每次生产者放到缓冲区的，就是一个数据单元；每次消费者从缓冲区取出的，也是一个数据单元。对于前一个帖子中寄信的例子，我们可以把每一封单独的信件看成是一个数据单元。

不过光这么介绍，太过于简单，无助于大伙儿分析出这玩意儿。所以，后面咱们来看一下数据单元需要具备哪些特性。搞明白这些特性之后，就容易从复杂的业务逻辑中分析出适合做数据单元的东西了。

数据单元的特性

分析数据单元，需要考虑如下几个方面的特性：

关联到业务对象

首先，数据单元必须关联到某种业务对象。在考虑该问题的时候，你必须深刻理解当前这个生产者／消费者模式所对应的业务逻辑，才能够作出合适的判断。

由于“寄信”这个业务逻辑比较简单，所以大伙儿很容易就可以判断出数据单元是啥。但现实生活中，往往没这么乐观。大多数业务逻辑都比较复杂，当中包含的业务对象是层次繁多、类型各异。在这种情况下，就不易作出决策了。

这一步很重要，如果选错了业务对象，会导致后续程序设计和编码实现的复杂度大为上升，增加了开发和维护成本。

完整性

所谓完整性，就是在传输过程中，要保证该数据单元的完整。要么整个数据单元被传递到消费者，要么完全没有传递到消费者。不允许出现部分传递的情形。

对于寄信来说，你不能把半封信放入邮筒；同样的，邮递员从邮筒中拿信，也不能只拿出信的一部分。

独立性

 所谓独立性，就是各个数据单元之间没有互相依赖，某个数据单元传输失败不应该影响已经完成传输的单元；也不应该影响尚未传输的单元。

为啥会出现传输失败捏？假如生产者的生产速度在一段时间内一直超过消费者的处理速度，那就会导致缓冲区不断增长并达到上限，之后的数据单元就会被丢弃。如果数据单元相互独立，等到生产者的速度降下来之后，后续的数据单元继续处理，不会受到牵连；反之，如果数据单元之间有某种耦合，导致被丢弃的数据单元会影响到后续其它单元的处理，那就会使程序逻辑变得非常复杂。

对于寄信来说，某封信弄丢了，不会影响后续信件的送达；当然更不会影响已经送达的信件。

颗粒度

前面提到，数据单元需要关联到某种业务对象。那么数据单元和业务对象是否要一一对应捏？很多场合确实是一一对应的。

不过，有时出于性能等因素的考虑，也可能会把N个业务对象打包成一个数据单元。那么，这个N该如何取值就是颗粒度的考虑了。颗粒度的大小是有讲究的。太大的颗粒度可能会造成某种浪费；太小的颗粒度可能会造成性能问题。颗粒度的权衡要基于多方面的因素，以及一些经验值的考量。

还是拿寄信的例子。如果颗粒度过小（比如设定为1），那邮递员每次只取出1封信。如果信件多了，那就得来回跑好多趟，浪费了时间。如果颗粒度太大（比如设定为100），那寄信的人得等到凑满100封信才拿去放入邮筒。假如平时很少写信，就得等上很久，也不太爽。

可能有同学会问：生产者和消费者的颗粒度能否设置成不同大小（比如对于寄信人设置成1，对于邮递员设置成100）。当然，理论上可以这么干，但是在某些情况下会增加程序逻辑和代码实现的复杂度。后面讨论具体技术细节时，或许会聊到这个问题。

作者：生产者/消费者模式的理解及实现（整理） - Luego - 博客园 (cnblogs.com)

生产者消费者问题是研究多线程程序时绕不开的经典问题之一，它描述是有一块缓冲区作为仓库，生产者可以将产品放入仓库，消费者则可以从仓库中取走产品。解决生产者/消费者问题的方法可分为两类：

（1）采用某种机制保护生产者和消费者之间的同步；

（2）在生产者和消费者之间建立一个管道。

第一种方式有较高的效率，并且易于实现，代码的可控制性较好，属于常用的模式。

第二种管道缓冲区不易控制，被传输数据对象不易于封装等，实用性不强。因此本文只介绍同步机制实现的生产者/消费者问题。

同步问题核心在于：如何保证同一资源被多个线程并发访问时的完整性。常用的同步方法是采用信号或加锁机制，保证资源在任意时刻至多被一个线程访问。Java语言在多线程编程上实现了完全对象化，提供了对同步机制的良好支持。在Java中一共有四种方法支持同步，其中前三个是同步方法，一个是管道方法。

（1）wait() / notify()方法

（2）await() / signal()方法

（3）BlockingQueue阻塞队列方法

（4）PipedInputStream / PipedOutputStream

本文只介绍最常用的前三种，第四种暂不做讨论，有兴趣的读者可以自己去网上找答案。

一、wait() / notify()方法

wait() / nofity()方法是基类Object的两个方法，也就意味着所有Java类都会拥有这两个方法，这样，我们就可以为任何对象实现同步机制。

wait()方法：当缓冲区已满/空时，生产者/消费者线程停止自己的执行，放弃锁，使自己处于等等状态，让其他线程执行。

notify()方法：当生产者/消费者向缓冲区放入/取出一个产品时，向其他等待的线程发出可执行的通知，同时放弃锁，使自己处于等待状态。

光看文字可能不太好理解，咱来段代码就明白了：

import java.util.LinkedList;
​
/**
 * 仓库类Storage实现缓冲区
 * 
 * Email:530025983@qq.com
 * 
 * @author MONKEY.D.MENG 2011-03-15
 * 
 */
public class Storage
{
    // 仓库最大存储量
    private final int MAX_SIZE = 100;
​
    // 仓库存储的载体
    private LinkedList<Object> list = new LinkedList<Object>();
​
    // 生产num个产品
    public void produce(int num)
    {
        // 同步代码段
        synchronized (list)
        {
            // 如果仓库剩余容量不足
            while (list.size() + num > MAX_SIZE)
            {
                System.out.println("【要生产的产品数量】:" + num + "/t【库存量】:"+ list.size() + "/t暂时不能执行生产任务!");
                try
                {
                    // 由于条件不满足，生产阻塞
                    list.wait();
                }
                catch (InterruptedException e)
                {
                    e.printStackTrace();
                }
            }
​
            // 生产条件满足情况下，生产num个产品
            for (int i = 1; i <= num; ++i)
            {
                list.add(new Object());
            }
​
            System.out.println("【已经生产产品数】:" + num + "/t【现仓储量为】:" + list.size());
​
            list.notifyAll();
        }
    }
​
    // 消费num个产品
    public void consume(int num)
    {
        // 同步代码段
        synchronized (list)
        {
            // 如果仓库存储量不足
            while (list.size() < num)
            {
                System.out.println("【要消费的产品数量】:" + num + "/t【库存量】:" + list.size() + "/t暂时不能执行生产任务!");
                try
                {
                    // 由于条件不满足，消费阻塞
                    list.wait();
                }
                catch (InterruptedException e)
                {
                    e.printStackTrace();
                }
            }
​
            // 消费条件满足情况下，消费num个产品
            for (int i = 1; i <= num; ++i)
            {
                list.remove();
            }
​
            System.out.println("【已经消费产品数】:" + num + "/t【现仓储量为】:" + list.size());
​
            list.notifyAll();
        }
    }
​
    // get/set方法
    public LinkedList<Object> getList()
    {
        return list;
    }
​
    public void setList(LinkedList<Object> list)
    {
        this.list = list;
    }
​
    public int getMAX_SIZE()
    {
        return MAX_SIZE;
    }
}
/**
 * 生产者类Producer继承线程类Thread
 * 
 * Email:530025983@qq.com
 * 
 * @author MONKEY.D.MENG 2011-03-15
 * 
 */
public class Producer extends Thread
{
    // 每次生产的产品数量
    private int num;
​
    // 所在放置的仓库
    private Storage storage;
​
    // 构造函数，设置仓库
    public Producer(Storage storage)
    {
        this.storage = storage;
    }
​
    // 线程run函数
    public void run()
    {
        produce(num);
    }
​
    // 调用仓库Storage的生产函数
    public void produce(int num)
    {
        storage.produce(num);
    }
​
    // get/set方法
    public int getNum()
    {
        return num;
    }
​
    public void setNum(int num)
    {
        this.num = num;
    }
​
    public Storage getStorage()
    {
        return storage;
    }
​
    public void setStorage(Storage storage)
    {
        this.storage = storage;
    }
}
/**
 * 消费者类Consumer继承线程类Thread
 * 
 * Email:530025983@qq.com
 * 
 * @author MONKEY.D.MENG 2011-03-15
 * 
 */
public class Consumer extends Thread
{
    // 每次消费的产品数量
    private int num;
​
    // 所在放置的仓库
    private Storage storage;
​
    // 构造函数，设置仓库
    public Consumer(Storage storage)
    {
        this.storage = storage;
    }
​
    // 线程run函数
    public void run()
    {
        consume(num);
    }
​
    // 调用仓库Storage的生产函数
    public void consume(int num)
    {
        storage.consume(num);
    }
​
    // get/set方法
    public int getNum()
    {
        return num;
    }
​
    public void setNum(int num)
    {
        this.num = num;
    }
​
    public Storage getStorage()
    {
        return storage;
    }
​
    public void setStorage(Storage storage)
    {
        this.storage = storage;
    }
}
/**
 * 测试类Test
 * 
 * Email:530025983@qq.com
 * 
 * @author MONKEY.D.MENG 2011-03-15
 * 
 */
public class Test
{
    public static void main(String[] args)
    {
        // 仓库对象
        Storage storage = new Storage();
​
        // 生产者对象
        Producer p1 = new Producer(storage);
        Producer p2 = new Producer(storage);
        Producer p3 = new Producer(storage);
        Producer p4 = new Producer(storage);
        Producer p5 = new Producer(storage);
        Producer p6 = new Producer(storage);
        Producer p7 = new Producer(storage);
​
        // 消费者对象
        Consumer c1 = new Consumer(storage);
        Consumer c2 = new Consumer(storage);
        Consumer c3 = new Consumer(storage);
​
        // 设置生产者产品生产数量
        p1.setNum(10);
        p2.setNum(10);
        p3.setNum(10);
        p4.setNum(10);
        p5.setNum(10);
        p6.setNum(10);
        p7.setNum(80);
​
        // 设置消费者产品消费数量
        c1.setNum(50);
        c2.setNum(20);
        c3.setNum(30);
​
        // 线程开始执行
        c1.start();
        c2.start();
        c3.start();
        p1.start();
        p2.start();
        p3.start();
        p4.start();
        p5.start();
        p6.start();
        p7.start();
    }
}
【要消费的产品数量】:50   【库存量】:0 暂时不能执行生产任务!
【要消费的产品数量】:30   【库存量】:0 暂时不能执行生产任务!
【要消费的产品数量】:20   【库存量】:0 暂时不能执行生产任务!
【已经生产产品数】:10    【现仓储量为】:10
【要消费的产品数量】:20   【库存量】:10    暂时不能执行生产任务!
【要消费的产品数量】:30   【库存量】:10    暂时不能执行生产任务!
【要消费的产品数量】:50   【库存量】:10    暂时不能执行生产任务!
【已经生产产品数】:10    【现仓储量为】:20
【要消费的产品数量】:50   【库存量】:20    暂时不能执行生产任务!
【要消费的产品数量】:30   【库存量】:20    暂时不能执行生产任务!
【已经消费产品数】:20    【现仓储量为】:0
【已经生产产品数】:10    【现仓储量为】:10
【已经生产产品数】:10    【现仓储量为】:20
【已经生产产品数】:80    【现仓储量为】:100
【要生产的产品数量】:10   【库存量】:100   暂时不能执行生产任务!
【已经消费产品数】:30    【现仓储量为】:70
【已经消费产品数】:50    【现仓储量为】:20
【已经生产产品数】:10    【现仓储量为】:30
【已经生产产品数】:10    【现仓储量为】:40

看完上述代码，对wait() / notify()方法实现的同步有了了解。你可能会对Storage类中为什么要定义public void produce(int num);和public void consume(int num);方法感到不解，为什么不直接在生产者类Producer和消费者类Consumer中实现这两个方法，却要调用Storage类中的实现呢？淡定，后文会有解释。我们先往下走。

练习一个小例子

使用多线程进行交替输出奇数偶数。t1输出奇数，t2输出偶数。

package com.example;
​
public class ProducerAndCustomer {
    /*
        使用多线程进行交替输出奇数偶数。t1输出奇数，t2输出偶数。
 */
    public static void main(String[] args) {
        Num num = new Num();
        Thread t1 = new Thread(new Producterb(num));
        Thread t2 = new Thread(new Customer(num));
        t1.setName("t1");
        t2.setName("t2");
        t1.start();
        t2.start();
​
    }
}
    class Num{
        int i = 1;
    }
    class Producterb implements Runnable{
        private Num num;
​
​
        public Producterb(Num num) {
            this.num = num;
        }
​
        @Override
        public void run() {
            while(true){
                synchronized (num) {
                    if (num.i % 2 == 0) {
                        try {
                            Thread.sleep(1000);//休眠一秒可以跟好的看见执行效果
                            num.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
​
                //程序可以走到这里说明num是奇数
                System.out.println(Thread.currentThread().getName()+"--->"+num.i++);
                    num.notifyAll();
            }
            }
        }
    }
    class Customer implements Runnable{
        private Num num;
​
        public Customer(Num num) {
            this.num = num;
        }
​
        @Override
        public void run() {
            while(true){
                synchronized (num) {
                    if (num.i % 2 == 1) {
                        try {
                            Thread.sleep(1000);
                            num.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
​
                //程序执行到此说明num是奇数
                System.out.println(Thread.currentThread().getName()+"--->"+num.i++);
​
                num.notifyAll();
            }
            }
        }
    }

二、await() / signal()方法

在JDK5.0之后，Java提供了更加健壮的线程处理机制，包括同步、锁定、线程池等，它们可以实现更细粒度的线程控制。await()和signal()就是其中用来做同步的两种方法，它们的功能基本上和wait() / nofity()相同，完全可以取代它们，但是它们和新引入的锁定机制Lock直接挂钩，具有更大的灵活性。通过在Lock对象上调用newCondition()方法，将条件变量和一个锁对象进行绑定，进而控制并发程序访问竞争资源的安全。下面来看代码：

import java.util.LinkedList;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
​
/**
 * 仓库类Storage实现缓冲区
 * 
 * Email:530025983@qq.com
 * 
 * @author MONKEY.D.MENG 2011-03-15
 * 
 */
public class Storage
{
    // 仓库最大存储量
    private final int MAX_SIZE = 100;
​
    // 仓库存储的载体
    private LinkedList<Object> list = new LinkedList<Object>();
​
    // 锁
    private final Lock lock = new ReentrantLock();
​
    // 仓库满的条件变量
    private final Condition full = lock.newCondition();
​
    // 仓库空的条件变量
    private final Condition empty = lock.newCondition();
​
    // 生产num个产品
    public void produce(int num)
    {
        // 获得锁
        lock.lock();
​
        // 如果仓库剩余容量不足
        while (list.size() + num > MAX_SIZE)
        {
            System.out.println("【要生产的产品数量】:" + num + "/t【库存量】:" + list.size()
                    + "/t暂时不能执行生产任务!");
            try
            {
                // 由于条件不满足，生产阻塞
                full.await();
            }
            catch (InterruptedException e)
            {
                e.printStackTrace();
            }
        }
​
        // 生产条件满足情况下，生产num个产品
        for (int i = 1; i <= num; ++i)
        {
            list.add(new Object());
        }
​
        System.out.println("【已经生产产品数】:" + num + "/t【现仓储量为】:" + list.size());
​
        // 唤醒其他所有线程
        full.signalAll();
        empty.signalAll();
​
        // 释放锁
        lock.unlock();
    }
​
    // 消费num个产品
    public void consume(int num)
    {
        // 获得锁
        lock.lock();
​
        // 如果仓库存储量不足
        while (list.size() < num)
        {
            System.out.println("【要消费的产品数量】:" + num + "/t【库存量】:" + list.size()
                    + "/t暂时不能执行生产任务!");
            try
            {
                // 由于条件不满足，消费阻塞
                empty.await();
            }
            catch (InterruptedException e)
            {
                e.printStackTrace();
            }
        }
​
        // 消费条件满足情况下，消费num个产品
        for (int i = 1; i <= num; ++i)
        {
            list.remove();
        }
​
        System.out.println("【已经消费产品数】:" + num + "/t【现仓储量为】:" + list.size());
​
        // 唤醒其他所有线程
        full.signalAll();
        empty.signalAll();
​
        // 释放锁
        lock.unlock();
    }
​
    // set/get方法
    public int getMAX_SIZE()
    {
        return MAX_SIZE;
    }
​
    public LinkedList<Object> getList()
    {
        return list;
    }
​
    public void setList(LinkedList<Object> list)
    {
        this.list = list;
    }
}
【要消费的产品数量】:50   【库存量】:0 暂时不能执行生产任务!
【要消费的产品数量】:30   【库存量】:0 暂时不能执行生产任务!
【已经生产产品数】:10    【现仓储量为】:10
【已经生产产品数】:10    【现仓储量为】:20
【要消费的产品数量】:50   【库存量】:20    暂时不能执行生产任务!
【要消费的产品数量】:30   【库存量】:20    暂时不能执行生产任务!
【已经生产产品数】:10    【现仓储量为】:30
【要消费的产品数量】:50   【库存量】:30    暂时不能执行生产任务!
【已经消费产品数】:20    【现仓储量为】:10
【已经生产产品数】:10    【现仓储量为】:20
【要消费的产品数量】:30   【库存量】:20    暂时不能执行生产任务!
【已经生产产品数】:80    【现仓储量为】:100
【要生产的产品数量】:10   【库存量】:100   暂时不能执行生产任务!
【已经消费产品数】:50    【现仓储量为】:50
【已经生产产品数】:10    【现仓储量为】:60
【已经消费产品数】:30    【现仓储量为】:30
【已经生产产品数】:10    【现仓储量为】:40
​

三、BlockingQueue阻塞队列方法

BlockingQueue是JDK5.0的新增内容，它是一个已经在内部实现了同步的队列，实现方式采用的是我们第2种await() / signal()方法。它可以在生成对象时指定容量大小。它用于阻塞操作的是put()和take()方法。

put()方法：类似于我们上面的生产者线程，容量达到最大时，自动阻塞。

take()方法：类似于我们上面的消费者线程，容量为0时，自动阻塞。

关于BlockingQueue的内容网上有很多，大家可以自己搜，我在这不多介绍。下面直接看代码，跟以往一样，我们只需要更改仓库类Storage的代码即可：

import java.util.concurrent.LinkedBlockingQueue;
​
/**
 * 仓库类Storage实现缓冲区
 * 
 */
public class Storage
{
    // 仓库最大存储量
    private final int MAX_SIZE = 100;
​
    // 仓库存储的载体
    private LinkedBlockingQueue<Object> list = new LinkedBlockingQueue<Object>(
            100);
​
    // 生产num个产品
    public void produce(int num)
    {
        // 如果仓库剩余容量为0
        if (list.size() == MAX_SIZE)
        {
            System.out.println("【库存量】:" + MAX_SIZE + "/t暂时不能执行生产任务!");
        }
​
        // 生产条件满足情况下，生产num个产品
        for (int i = 1; i <= num; ++i)
        {
            try
            {
                // 放入产品，自动阻塞
                list.put(new Object());
            }
            catch (InterruptedException e)
            {
                e.printStackTrace();
            }
​
            System.out.println("【现仓储量为】:" + list.size());
        }
    }
​
    // 消费num个产品
    public void consume(int num)
    {
        // 如果仓库存储量不足
        if (list.size() == 0)
        {
            System.out.println("【库存量】:0/t暂时不能执行生产任务!");
        }
​
        // 消费条件满足情况下，消费num个产品
        for (int i = 1; i <= num; ++i)
        {
            try
            {
                // 消费产品，自动阻塞
                list.take();
            }
            catch (InterruptedException e)
            {
                e.printStackTrace();
            }
        }
​
        System.out.println("【现仓储量为】:" + list.size());
    }
​
    // set/get方法
    public LinkedBlockingQueue<Object> getList()
    {
        return list;
    }
​
    public void setList(LinkedBlockingQueue<Object> list)
    {
        this.list = list;
    }
​
    public int getMAX_SIZE()
    {
        return MAX_SIZE;
    }
}
【库存量】:0 暂时不能执行生产任务!
【库存量】:0 暂时不能执行生产任务!
【现仓储量为】:1
【现仓储量为】:1
【现仓储量为】:3
【现仓储量为】:4
【现仓储量为】:5
【现仓储量为】:6
【现仓储量为】:7
【现仓储量为】:8
【现仓储量为】:9
【现仓储量为】:10
【现仓储量为】:11
【现仓储量为】:1
【现仓储量为】:2
【现仓储量为】:13
【现仓储量为】:14
【现仓储量为】:17
【现仓储量为】:19
【现仓储量为】:20
【现仓储量为】:21
【现仓储量为】:22
【现仓储量为】:23
【现仓储量为】:24
【现仓储量为】:25
【现仓储量为】:26
【现仓储量为】:12
【现仓储量为】:1
【现仓储量为】:1
【现仓储量为】:2
【现仓储量为】:3
【现仓储量为】:4
【现仓储量为】:5
【现仓储量为】:6
【现仓储量为】:7
【现仓储量为】:27
【现仓储量为】:8
【现仓储量为】:6
【现仓储量为】:18
【现仓储量为】:2
【现仓储量为】:3
【现仓储量为】:4
【现仓储量为】:5
【现仓储量为】:6
【现仓储量为】:7
【现仓储量为】:8
【现仓储量为】:9
【现仓储量为】:10
【现仓储量为】:16
【现仓储量为】:11
【现仓储量为】:12
【现仓储量为】:13
【现仓储量为】:14
【现仓储量为】:15
【现仓储量为】:1
【现仓储量为】:2
【现仓储量为】:3
【现仓储量为】:3
【现仓储量为】:15
【现仓储量为】:1
【现仓储量为】:0
【现仓储量为】:1
【现仓储量为】:1
【现仓储量为】:1
【现仓储量为】:2
【现仓储量为】:3
【现仓储量为】:4
【现仓储量为】:0
【现仓储量为】:1
【现仓储量为】:5
【现仓储量为】:6
【现仓储量为】:7
【现仓储量为】:8
【现仓储量为】:9
【现仓储量为】:10
【现仓储量为】:11
【现仓储量为】:12
【现仓储量为】:13
【现仓储量为】:14
【现仓储量为】:15
【现仓储量为】:16
【现仓储量为】:17
【现仓储量为】:1
【现仓储量为】:1
【现仓储量为】:2
【现仓储量为】:3
【现仓储量为】:4
【现仓储量为】:5
【现仓储量为】:6
【现仓储量为】:3
【现仓储量为】:3
【现仓储量为】:1
【现仓储量为】:2
【现仓储量为】:3
【现仓储量为】:4
【现仓储量为】:5
【现仓储量为】:6
【现仓储量为】:7
【现仓储量为】:8
【现仓储量为】:9
【现仓储量为】:10
【现仓储量为】:11
【现仓储量为】:12
【现仓储量为】:13
【现仓储量为】:14
【现仓储量为】:15
【现仓储量为】:16
【现仓储量为】:17
【现仓储量为】:18
【现仓储量为】:19
【现仓储量为】:6
【现仓储量为】:7
【现仓储量为】:8
【现仓储量为】:9
【现仓储量为】:10
【现仓储量为】:11
【现仓储量为】:12
【现仓储量为】:13
【现仓储量为】:14
【现仓储量为】:15
【现仓储量为】:16
【现仓储量为】:17
【现仓储量为】:18
【现仓储量为】:19
【现仓储量为】:20
【现仓储量为】:21
【现仓储量为】:22
【现仓储量为】:23
【现仓储量为】:24
【现仓储量为】:25
【现仓储量为】:26
【现仓储量为】:27
【现仓储量为】:28
【现仓储量为】:29
【现仓储量为】:30
【现仓储量为】:31
【现仓储量为】:32
【现仓储量为】:33
【现仓储量为】:34
【现仓储量为】:35
【现仓储量为】:36
【现仓储量为】:37
【现仓储量为】:38
【现仓储量为】:39
【现仓储量为】:40

当然，你会发现这时对于public void produce(int num);和public void consume(int num);方法业务逻辑上的实现跟前面两个例子不太一样，没关系，这个例子只是为了说明BlockingQueue阻塞队列的使用。

有时使用BlockingQueue可能会出现put()和System.out.println()输出不匹配的情况，这是由于它们之间没有同步造成的。当缓冲区已满，生产者在put()操作时，put()内部调用了await()方法，放弃了线程的执行，然后消费者线程执行，调用take()方法，take()内部调用了signal()方法，通知生产者线程可以执行，致使在消费者的println()还没运行的情况下生产者的println()先被执行，所以有了输出不匹配的情况。

对于BlockingQueue大家可以放心使用，这可不是它的问题，只是在它和别的对象之间的同步有问题。

热门相关：超武穿梭   豪门闪婚：帝少的神秘冷妻   豪门闪婚：帝少的神秘冷妻   豪门闪婚：帝少的神秘冷妻   天启预报