线程安全的生产者消费者四种实现方法
- 2019 年 11 月 12 日
- 筆記
问题描述
在IT技术面试过程中，我们经常会遇到生产者消费者问题(Producer-consumer problem)， 这是多线程并发协作问题的经典案例。场景中包含三个对象，生产者(Producer)，消费者(Consumer)以及一个固定大小的缓冲区(Buffer)。生产者的主要作用是不断生成数据放到缓冲区，消费者则从缓冲区不断消耗数据。该问题的关键是如何线程安全的操作共享数据块，保证生产者线程和消费者线程可以正确的更新数据块，主要考虑 1. 生产者不会在缓冲区满时加入数据. 2. 消费者应当停止在缓冲区时消耗数据. 3. 在同一时间应当只允许一个生产者或者消费者访问共享缓冲区（这一点是对于互斥操作访问共享区块的要求）。
解决方案
解决问题以上问题通常有信号量，wait & notify, 管道或者阻塞队列等几种思路。本文以Java语言为例一一进行举例讲解。
信号量
信号量(Semaphore)也称信号灯，是用来控制资源被同时访问的个数，比如控制访问数据库最大连接数的数量，线程通过acquire()获得连接许可，完成数据操作后，通过release()释放许可。对于生产者消费者问题来说，为了满足线程安全操作的要求，同一时间我们只允许一个线程访问共享数据区，因此需要一个大小为1的信号量mutex来控制互斥操作。注意到我们还定义了notFull 和 notEmpty 信号量，notFull用于标识当前可用区块的空间大小，当notFull size 大于0时表明"not full", producer 可以继续生产，等于0时表示空间已满，无法继续生产；同样，对于notEmpty信号量来说，大于0时表明 "not empty", consumer可以继续消耗，等于0 时表明没有产品，无法继续消耗。notFull初始size 为5 (5个available空间可供生产)，notEmpty初始为0(没有产品可供消耗)。
/*** 数据仓储class，所有的producer和consumer共享这个class对象 **/ static class DataWareHouse { //共享数据区 private final Queue<String> data = new LinkedList(); //非满锁 private final Semaphore notFull; //非空锁 private final Semaphore notEmpty; //互斥锁 private final Semaphore mutex; public DataWareHouse(int capacity) { this.notFull = new Semaphore(capacity); this.notEmpty = new Semaphore(0); mutex = new Semaphore(1); } public void offer(String x) throws InterruptedException { notFull.acquire(); //producer获取信号，notFull信号量减一 mutex.acquire(); //当前进程获得信号，mutex信号量减1，其他线程被阻塞操作共享区块data data.add(x); mutex.release(); //mutex信号量+1, 其他线程可以继续信号操作共享区块data notEmpty.release(); //成功生产数据，notEmpty信号量加1 } public String poll() throws InterruptedException { notEmpty.acquire(); //notEmpty信号减一 mutex.acquire(); String result = data.poll(); mutex.release(); notFull.release(); //成功消耗数据, notFull信号量加1 return result; } } /**Producer线程**/ static class Producer implements Runnable { private final DataWareHouse dataWareHouse; public Producer(final DataWareHouse dataWareHouse) { this.dataWareHouse = dataWareHouse; } @Override public void run() { while (true) { try { Thread.sleep(100); //生产的速度慢于消耗的速率 String s = UUID.randomUUID().toString(); System.out.println("put data " + s); dataWareHouse.offer(s); } catch (InterruptedException e) { e.printStackTrace(); } } } } /**Consumer线程**/ static class Consumer implements Runnable { private final DataWareHouse dataWareHouse; public Consumer(final DataWareHouse dataWareHouse) { this.dataWareHouse = dataWareHouse; } @Override public void run() { while (true) { while (true) { try { System.out.println("get data " + dataWareHouse.poll()); } catch (InterruptedException e) { e.printStackTrace(); } } } } } //测试代码 public static void main(String[] args) { final DataWareHouse dataWareHouse = new DataWareHouse(5); //三个producer 持续生产 for (int i = 0; i < 3; i++) { Thread t = new Thread(new Producer(dataWareHouse)); t.start(); } //三个consumer 持续消耗 for (int i = 0; i < 3; i++) { Thread t = new Thread(new Consumer(dataWareHouse)); t.start(); } }Wait 和 Notify 机制
Java Object对象类中包含三个final methods来允许线程之间进行通信，告知资源的状态。它们分别是wait(), notify(), 和notifyAll()。
wait(): 顾名思义告诉当前线程释放锁，陷入休眠状态(waiting状态)，等待资源。wait 方法本身是一个native method，它在Java中的使用语法如下所示:
synchronized(lockObject ) { while( ! condition ) { lockObject.wait(); } //take the action here; }notify(): 用于唤醒waiting状态的线程, 同时释放锁，被唤醒的线程可以重新获得锁访问资源。它的基本语法 如下
synchronized(lockObject) { //establish_the_condition; lockObject.notify(); //any additional code if needed }notifyAll(): 不同于notify()，它用于唤醒所有处于waiting状态的线程。语法如下:
synchronized(lockObject) { establish_the_condition; lockObject.notifyAll(); }说完了这三个方法，来看下如何使用wait & notify(All) 来解决我们的问题。新的DataWareHouse 类如下所示：
//producer类和consumer共享对象 static class DataWareHouse { //共享数据区 private final Queue<String> data = new LinkedList(); private int capacity; private int size = 0; public DataWareHouse(int capacity) { this.capacity = capacity; } public synchronized void offer(String x) throws InterruptedException { while (size == capacity) { //当buffer满时，producer进入waiting 状态 this.wait(); //使用this对象来加锁 } data.add(x); size++; notifyAll(); //当buffer 有数据时，唤醒所有等待的consumer线程 } public synchronized String poll() throws InterruptedException { while (size == 0) {//当buffer为空时，consumer 进入等待状态 this.wait(); } String result = data.poll(); size--; notifyAll(); //当数据被消耗，空间被释放，通知所有等待的producer。 return result; } }Note: 在方法上使用synchronized 等价于在方法体内使用synchronized(this)，两者都是使用this对象作为锁。
生产者和消费者类，以及测试代码和 信号量 section 相同，不做重复列举了。
管道
管道Pipe是实现进程或者线程（线程之间通常通过共享内存实现通讯，而进程则通过scoket，管道，消息队列等技术）之间通信常用方式，它连接输入流和输出流，基于生产者- 消费者模式构建的一种技术。具体实现可以通过创建一个管道输入流对象和管道输出流对象，然后将输入流和输出流就行链接，生产者通过往管道中写入数据，而消费者在管道数据流中读取数据，通过这种方式就实现了线程之间的互相通讯。
具体实现代码如下所示
public class PipeSolution { static class DataWareHouse implements Closeable { private final PipedInputStream pis; private final PipedOutputStream pos; public DataWareHouse() throws IOException { pis = new PipedInputStream(); pos = new PipedOutputStream(); pis.connect(pos); //连接管道 } //向管道中写入数据 public void offer(int val) throws IOException { pos.write(val); pos.flush(); } //从管道中取数据. public int poll() throws IOException { //当管道中没有数据，方法阻塞 return pis.read(); } //关闭管道 @Override public void close() throws IOException { if (pis != null) { pis.close(); } if (pos != null) { pos.close(); } } } //consumer类 static class Consumer implements Runnable { private final DataWareHouse dataWareHouse; Consumer(DataWareHouse dataWareHouse) { this.dataWareHouse = dataWareHouse; } @Override public void run() { try { //消费者不断从管道中读取数据 while (true) { int num = dataWareHouse.poll(); System.out.println("get data +" + num); } } catch (IOException e) { throw new RuntimeException(e); } } } static class Producer implements Runnable { private final DataWareHouse dataWareHouse; private final Random random = new Random(); Producer(DataWareHouse dataWareHouse) { this.dataWareHouse = dataWareHouse; } @Override public void run() { try { //生产者不断向管道中写入数据 while (true) { int num = random.nextInt(256); dataWareHouse.offer(num); System.out.println("put data +" + num); Thread.sleep(1000); } } catch (Exception e) { throw new RuntimeException(e); } } public static void main(String[] args) throws IOException { DataWareHouse dataWareHouse = new DataWareHouse(); new Thread(new Producer(dataWareHouse)).start(); new Thread(new Consumer(dataWareHouse)).start(); } }阻塞队列
阻塞队列(BlockingQueue)，具有1. 当队列满了的时候阻塞入队列操作 2. 当队列空了的时候阻塞出队列操作 3. 线程安全 的特性，因而阻塞队列通常被视为实现生产消费者模式最便捷的工具，其中DataWareHouse类实现代码如下：
static class DataWareHouse { //共享数据区 private final BlockingQueue<String> blockingQueue; public DataWareHouse(int capacity) { this.blockingQueue = new ArrayBlockingQueue<>(capacity); } public void offer(String x) { blockingQueue.offer(x); } public String poll() { return blockingQueue.poll(); } }生产者和消费者类，以及测试代码和 信号量 section 相同，在此不做重复列举了。
总结
生产者消费者问题是面试中经常会遇到的题目，本文总结了几种常见的实现方式，面试过程中通常不必要向面试官描述过多实现细节，说出每种实现方式的特点即可。希望能给大家带来帮助。
