从零学习JAVA多线程（四）：阻塞队列和生产者消费者模式

阻塞队列
阻塞队列

阻塞队列的方法

几种阻塞队列的实现

生产者和消费者模式
什么是生产者消费者模式

生产者消费者模式的优点

生产者消费者模式的实现
waitnotify实现

awaitsignal实现

阻塞队列实现

在很多书和博客里，阻塞队列和生产者消费者模式是绑在一起的，好像二者关系是绝对密不可分互为实现方案的，但事实上并不是，生产者消费者模式是多线程编程中常见的解决方案，阻塞队列则是实现生产者消费者方案的一种（较好的）实现方法。

我们就阻塞队列开始讲，然后在生产者和消费者模式的讲解中会给出几种常见的实现方案。

阻塞队列

阻塞队列

阻塞队列（BlockingQueue）是线程安全版本的队列，它支持线程阻塞。当我们向一个空的阻塞队列请求数据的时候，它会阻塞直至有新的数据插入；相对的，如果我们向一个全满的队列中插入数据，它也会阻塞知道有新的位置可供数据插入。

阻塞队列的方法

在阻塞队列中存在以下方法可供使用：

方法	正常动作	特殊处理
add	添加一个元素	如果队列满，抛出IllegalStatException异常
element	返回队列的头元素	如果队列空，抛出NoSuchElementException异常
remove	移除并返回队列的头元素	如果队列空，抛出NoSuchElementException异常
offer	添加一个元素并返回true	如果队列满，返回false
peek	返回队列的头元素	如果队列为空，返回null
poll	移除并返回队列的头元素	如果队列为空，返回null
put	添加一个元素到队列	如果队列满，则阻塞
take	移除并返回队列的头元素	如果队列空，则阻塞

从上面的方法集合里可以看出，BlockingQueue是支持阻塞动作的。

几种阻塞队列的实现

在Java的java.util.concurrent包中，提供了六种Blocking的实现：

LinkedBlockingQueue:由链表结构组成的有界阻塞队列，如果不指定最大容量，则LinkedBlockingQueue是没有上边界的（注意支持的最大长度是 Integer.MAX_VALUE，也可以认为是长度为 Integer.MAX_VALUE的有界队列，只是不需要声明长度）

ArrayBlockingQueue:由数组结构组成的有界阻塞队列。在构造时需要指明容量，还有一个可选参数来设定公平性。如果设定了公平性参数，那么等待时间最长的线程会得到优先处理（会降低性能）

PriorityBlockingQueue:支持优先级排序的无界阻塞队列。队列中的元素会按照优先级顺序被移出。

DelayQueue:只有延时期满才能取出数据的无界阻塞队列。

SynchronousQueue:一个不存储元素的阻塞队列。只有当请求元素时，才允许添加。也就是想要put，必须要对应的take。

LinkedTransferQueue(Java 7增加):基于链表的无界阻塞队列。采用预占模式分配。当有线程来请求元素时，如果队列中有直接拿走，如果没有就标记一下开始等，等到了就走。

生产者和消费者模式

什么是生产者消费者模式

所谓生产者就是一个生产数据（指令）的线程，消费者就是一个消费数据（指令）的线程。生产者不断地产生数据，消费者不断地使用数据，像很多设计模式一样，为了二者的解耦，我们会生成一个缓冲区，这个缓冲区负责接收和推出数据，并且处理数据缺少或者溢出时的特殊情况。

为了形象化这个场景，我们可以设想一下饭店里吃饭做饭的场景：

顾客不断走进来需求新的食物，厨师们不断劳动生产食物。这就是生产者和消费者关系。一般情况下在饭点厨师的生产能力可能跟不上点餐的速度，就会有很多客户等待，这些客户里可能有人等的时间长，有人是熟客或者Vip（优先级高），情况就变得复杂起来。这时候，就出现了服务员（缓冲区），有了服务员，顾客和厨师就不用直接沟通了，厨师也不用记着谁先来谁后到，顾客也不用因为不晓得还要等多久换家店了。点菜上菜记录顺序和偏好都由服务员负责，整个事务场景就会一下子清晰起来。

生产者消费者模式的优点

生产者消费者模式作为一个具有普遍实用性的方案，具有以下的特点；

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

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

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

支持忙闲不均

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

生产者消费者模式的实现

生产者与消费者模式有几种的实现方案：

wait()/notify()实现

await()/signal()实现

阻塞队列实现

Semaphore实现（较为少见，不做介绍）

PipedInputStream / PipedOutputStream（较为少见，不做介绍）

wait()/notify()实现

复习一下wait()和notify()方法。

是Object的公用方法。

调用了wait()方法的线程进入等待状态，直到有别的线程用notify()或者notifyAll()方法唤醒它。被唤醒的线程在重新获得锁之后可以继续执行。

notify方法用于唤醒某个等待的线程。

在生产者消费者模式中的作用：

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

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

示例代码如下：

缓冲区：

package com.vivi.myTest.synchronizedTest.produceAndConsume;

import java.util.LinkedList;

/**
* Created by vivit on 2017/10/16.
*/
public class Storage {
//    仓库最大容量
private final int MAX_SIZE = 100;

//仓库的存储载体
private LinkedList<Object> list = new LinkedList<>();

// 生产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;
}
}

生产者：

package com.vivi.myTest.synchronizedTest.produceAndConsume;

/**
* Created by vivit on 2017/10/16.
*/
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;
}
}

消费者

package com.vivi.myTest.synchronizedTest.produceAndConsume;

/**
* Created by vivit on 2017/10/16.
*/
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;
}
}

测试入口：

package com.vivi.myTest.synchronizedTest.produceAndConsume;

/**
* Created by vivit on 2017/10/16.
*/
public class TestEntrance {
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/t【库存量】:0/t暂时不能执行生产任务!
【要消费的产品数量】:20/t【库存量】:0/t暂时不能执行生产任务!
【已经生产产品数】:10/t【现仓储量为】:10
【要消费的产品数量】:30/t【库存量】:10/t暂时不能执行生产任务!
【要消费的产品数量】:20/t【库存量】:10/t暂时不能执行生产任务!
【要消费的产品数量】:50/t【库存量】:10/t暂时不能执行生产任务!
【已经生产产品数】:10/t【现仓储量为】:20
【已经生产产品数】:10/t【现仓储量为】:30
【已经生产产品数】:10/t【现仓储量为】:40
【要消费的产品数量】:50/t【库存量】:40/t暂时不能执行生产任务!
【已经消费产品数】:20/t【现仓储量为】:20
【要消费的产品数量】:30/t【库存量】:20/t暂时不能执行生产任务!
【已经生产产品数】:80/t【现仓储量为】:100
【已经消费产品数】:50/t【现仓储量为】:50
【已经生产产品数】:10/t【现仓储量为】:60
【已经生产产品数】:10/t【现仓储量为】:70
【已经消费产品数】:30/t【现仓储量为】:40

await()/signal()实现

因为缓冲区解耦了生产者和消费者的关系，当我们想要替换掉wait()/notify()方法时，只需要修改Storage即可，不需要修改生产者和消费者类。

代码如下：

package com.vivi.myTest.synchronizedTest.produceAndConsume;

import java.util.LinkedList;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
* Created by vivit on 2017/10/16.
*/
public class StorageWithAwait {
// 仓库最大存储量
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

阻塞队列实现

这里就是我们的重点部分了，就是使用阻塞队列实现生产者消费者模式。

根据第一部分对于阻塞队列的介绍，我们再次改写Storage类。

package com.vivi.myTest.synchronizedTest.produceAndConsume;

import java.util.concurrent.LinkedBlockingQueue;

/**
* Created by vivit on 2017/10/16.
*/
public class StorageWithBlockingQueue {
// 仓库最大存储量
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
.....

注：文中代码引用了MONKEY_D_MENG作者的演示代码生产者/消费者问题的多种Java实现方式

到这里，阻塞线程和生产者消费者模式就都讲清楚了，自己也难得系统复习一下多线程的基础知识。下一篇文章会力争把线程池讲清楚。

相对于最开始的两篇文章，生产者和消费者模式开始，才开始真正有实践意义，前面的基础知识只能让我们同时多写几个“Hello world”。

碎碎念几下，我觉得线程池中这个“池”的概念真的很重要很基础，基本是一个程序员的必知必会内容。我陪boss面试面别人的时候，线程池数据库连接池各种资源池基本是必问内容。我自己工作时也确实体会到了池子的重要性，希望下一个文章能把池子的概念和线程池梳理清楚。

附上版权声明：

原作者：Vi_error，博客地址：Vi_error.nextval

转载请保持署名和注明原地址**