一文吃透生產(chǎn)者和消費者模型！
作者：志哥聊技術
簡單的說，生產(chǎn)者和消費者之間不直接進行交互，而是通過一個緩沖區(qū)來進行交互，生產(chǎn)者負責生成數(shù)據(jù)，然后存入緩沖區(qū)；消費者則負責處理數(shù)據(jù)，從緩沖區(qū)獲取。
01、背景介紹
在 Java 多線程編程中，還有一個非常重要的設計模式，它就是：生產(chǎn)者和消費者模型。
這種模型可以充分發(fā)揮 cpu 的多線程特性，通過一些平衡手段能有效的提升系統(tǒng)整體處理數(shù)據(jù)的速度，減輕系統(tǒng)負載，提高程序的效率和穩(wěn)定性，同時實現(xiàn)模塊之間的解耦。
那什么是生產(chǎn)者和消費者模型呢？
簡單的說，生產(chǎn)者和消費者之間不直接進行交互，而是通過一個緩沖區(qū)來進行交互，生產(chǎn)者負責生成數(shù)據(jù)，然后存入緩沖區(qū)；消費者則負責處理數(shù)據(jù)，從緩沖區(qū)獲取。
大致流程圖如下：
圖片
對于最簡單的生產(chǎn)者和消費者模型，總結下來，大概有以下幾個特點：
- 緩沖區(qū)為空的時候，消費者不能消費，會進入休眠狀態(tài)，直到有新數(shù)據(jù)進入緩沖區(qū)，再次被喚醒
- 緩沖區(qū)填滿的時候，生產(chǎn)者不能生產(chǎn)，也會進入休眠狀態(tài)，直到緩沖區(qū)有空間，再次被喚醒
生產(chǎn)者和消費者模型作為一個非常重要的設計模型，它的優(yōu)點在于：
- 解耦：生產(chǎn)者和消費者之間不直接進行交互，即使生產(chǎn)者和消費者的代碼發(fā)生變化，也不會對對方產(chǎn)生影響
- 消峰：例如在某項工作中，假如 A 操作生產(chǎn)數(shù)據(jù)的速度很快，B 操作處理速度很慢，那么 A 操作就必須等待 B 操作完成才能結束，反之亦然。如果將 A 操作和B 操作進行解耦，中間插入一個緩沖區(qū)，這樣 A 操作將生產(chǎn)的數(shù)據(jù)存入緩沖區(qū)，就接受了；B 操作從緩沖區(qū)獲取數(shù)據(jù)并進行處理，平衡好 A 操作和 B 操作之間的緩沖區(qū)，可以顯著提升系統(tǒng)的數(shù)據(jù)處理能力
生產(chǎn)者和消費者模型的應用場景非常多，例如 Java 的線程池任務執(zhí)行框架、消息中間件 rabbitMQ 等，因此掌握生產(chǎn)者和消費者模型，對于開發(fā)者至關重要。
下面我們通過幾個案例，一起來了解一下生產(chǎn)者和消費者設計模型的實踐思路。
02、代碼實踐
2.1、利用 wait / notify 方法實現(xiàn)思路
生產(chǎn)者和消費者模型，最簡單的一種技術實踐方案就是基于線程的 wait() / notify() 方法，也就是通知和喚醒機制，可以將兩個操作實現(xiàn)解耦，具體代碼實踐如下。
/**
* 緩沖區(qū)容器類
*/
public class Container {
/**
* 緩沖區(qū)最大容量
*/
private int capacity = 3;
/**
* 緩沖區(qū)
*/
private LinkedList<Integer> list = new LinkedList<Integer>();
/**
* 添加數(shù)據(jù)到緩沖區(qū)
* @param value
*/
public synchronized void add(Integer value) {
if(list.size() >= capacity){
System.out.println("生產(chǎn)者："+ Thread.currentThread().getName()+"，緩沖區(qū)已滿,生產(chǎn)者進入waiting...");
try {
// 進入等待狀態(tài)
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println("生產(chǎn)者："+ Thread.currentThread().getName()+"，add：" + value);
list.add(value);
//喚醒其他所有處于wait()的線程，包括消費者和生產(chǎn)者
notifyAll();
}
/**
* 從緩沖區(qū)獲取數(shù)據(jù)
*/
public synchronized void get() {
if(list.size() == 0){
System.out.println("消費者："+ Thread.currentThread().getName()+"，緩沖區(qū)為空,消費者進入waiting...");
try {
// 進入等待狀態(tài)
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// 從頭部獲取數(shù)據(jù)，并移除元素
Integer val = list.removeFirst();
System.out.println("消費者："+ Thread.currentThread().getName()+"，value：" + val);
//喚醒其他所有處于wait()的線程，包括消費者和生產(chǎn)者
notifyAll();
}
}/**
* 生產(chǎn)者類
*/
public class Producer extends Thread{
private Container container;
public Producer(Container container) {
this.container = container;
}
@Override
public void run() {
for (int i = 0; i < 6; i++) {
container.add(i);
}
}
}/**
* 消費者類
*/
public class Consumer extends Thread{
private Container container;
public Consumer(Container container) {
this.container = container;
}
@Override
public void run() {
for (int i = 0; i < 6; i++) {
container.get();
}
}
}/**
* 測試類
*/
public class MyThreadTest {
public static void main(String[] args) {
Container container = new Container();
Producer producer = new Producer(container);
Consumer consumer = new Consumer(container);
producer.start();
consumer.start();
}
}運行結果如下：
生產(chǎn)者：Thread-0，add：0
生產(chǎn)者：Thread-0，add：1
生產(chǎn)者：Thread-0，add：2
生產(chǎn)者：Thread-0，緩沖區(qū)已滿,生產(chǎn)者進入waiting...
消費者：Thread-1，value：0
消費者：Thread-1，value：1
消費者：Thread-1，value：2
消費者：Thread-1，緩沖區(qū)為空,消費者進入waiting...
生產(chǎn)者：Thread-0，add：3
生產(chǎn)者：Thread-0，add：4
生產(chǎn)者：Thread-0，add：5
消費者：Thread-1，value：3
消費者：Thread-1，value：4
消費者：Thread-1，value：5從日志上可以很清晰的看到，生產(chǎn)者線程生產(chǎn)一批數(shù)據(jù)之后，當緩沖區(qū)已經(jīng)滿了，會進入等待狀態(tài)，此時會通知消費者線程；消費者線程處理完數(shù)據(jù)之后，當緩沖區(qū)沒有數(shù)據(jù)時，也會進入等待狀態(tài)，再次通知生產(chǎn)者線程。
2.2、利用 await / signal 方法實現(xiàn)思路
除此之外，我們還可以利用ReentrantLock和Condition類中的 await() / signal() 方法實現(xiàn)生產(chǎn)者和消費者模型。
緩沖區(qū)容器類，具體代碼實踐如下。
/**
* 緩沖區(qū)容器類
*/
public class Container {
private Lock lock = new ReentrantLock();
private Condition condition = lock.newCondition();
private int capacity = 3;
private LinkedList<Integer> list = new LinkedList<Integer>();
/**
* 添加數(shù)據(jù)到緩沖區(qū)
* @param value
*/
public void add(Integer value) {
boolean flag = false;
try {
flag = lock.tryLock(3, TimeUnit.SECONDS);
if(list.size() >= capacity){
System.out.println("生產(chǎn)者："+ Thread.currentThread().getName()+"，緩沖區(qū)已滿,生產(chǎn)者進入waiting...");
// 進入等待狀態(tài)
condition.await();
}
System.out.println("生產(chǎn)者："+ Thread.currentThread().getName()+"，add：" + value);
list.add(value);
//喚醒其他所有處于wait()的線程，包括消費者和生產(chǎn)者
condition.signalAll();
} catch (Exception e) {
e.printStackTrace();
} finally {
if(flag){
lock.unlock();
}
}
}
/**
* 從緩沖區(qū)獲取數(shù)據(jù)
*/
public void get() {
boolean flag = false;
try {
flag = lock.tryLock(3, TimeUnit.SECONDS);
if(list.size() == 0){
System.out.println("消費者："+ Thread.currentThread().getName()+"，緩沖區(qū)為空,消費者進入waiting...");
// 進入等待狀態(tài)
condition.await();
}
// 從頭部獲取數(shù)據(jù)，并移除元素
Integer val = list.removeFirst();
System.out.println("消費者："+ Thread.currentThread().getName()+"，value：" + val);
//喚醒其他所有處于wait()的線程，包括消費者和生產(chǎn)者
condition.signalAll();
} catch (Exception e) {
e.printStackTrace();
} finally {
if(flag){
lock.unlock();
}
}
}
}生產(chǎn)者、消費者、測試類代碼，跟上文一致，運行結果和上文介紹的也是一樣。
2.3、多生產(chǎn)者和消費者的實現(xiàn)思路
上面介紹的都是一個生產(chǎn)者線程和一個消費者線程，模型比較簡單。實際上，在業(yè)務開發(fā)中，經(jīng)常會出現(xiàn)多個生產(chǎn)者線程和多個消費者線程，按照以上的實現(xiàn)思路，會出現(xiàn)什么情況呢？
有可能會出現(xiàn)程序假死現(xiàn)象！下面我們來分析一下案例，假如有兩個生產(chǎn)者線程 a1、a2，兩個消費者線程 b1、b2，執(zhí)行過程如下：
- 1.生產(chǎn)者線程 a1 執(zhí)行生產(chǎn)數(shù)據(jù)的操作，發(fā)現(xiàn)緩沖區(qū)數(shù)據(jù)已經(jīng)填滿了，然后進入等待階段，同時向外發(fā)起通知，喚醒其它線程
- 2.因為線程喚醒具有隨機性，本應該喚醒消費者線程 b1，結果可能生產(chǎn)者線程 a2 被喚醒，檢查緩沖區(qū)數(shù)據(jù)已經(jīng)填滿了，又進入等待階段，緊接向外發(fā)起通知，消費者線程得不到被執(zhí)行的機會
- 3.消費者線程 b1、b2，也有可能會出現(xiàn)這個現(xiàn)象，本應該喚醒生產(chǎn)者線程，結果喚醒了消費者線程
遇到這種情況，應該如何解決呢？
因為ReentrantLock和Condition的結合，編程具有高度靈活性，我們可以采用這種組合解決多生產(chǎn)者和多消費者中的假死問題。
具體實現(xiàn)邏輯如下：
/**
* 緩沖區(qū)容器類
*/
public class ContainerDemo {
private Lock lock = new ReentrantLock();
private Condition producerCondition = lock.newCondition();
private Condition consumerCondition = lock.newCondition();
private int capacity = 3;
private LinkedList<Integer> list = new LinkedList<Integer>();
/**
* 添加數(shù)據(jù)到緩沖區(qū)
* @param value
*/
public void add(Integer value) {
boolean flag = false;
try {
flag = lock.tryLock(3, TimeUnit.SECONDS);
if(list.size() >= capacity){
System.out.println("生產(chǎn)者："+ Thread.currentThread().getName()+"，緩沖區(qū)已滿,生產(chǎn)者進入waiting...");
// 生產(chǎn)者進入等待狀態(tài)
producerCondition.await();
}
System.out.println("生產(chǎn)者："+ Thread.currentThread().getName()+"，add：" + value);
list.add(value);
// 喚醒所有消費者處于wait()的線程
consumerCondition.signalAll();
} catch (Exception e) {
e.printStackTrace();
} finally {
if(flag){
lock.unlock();
}
}
}
/**
* 從緩沖區(qū)獲取數(shù)據(jù)
*/
public void get() {
boolean flag = false;
try {
flag = lock.tryLock(3, TimeUnit.SECONDS);
if(list.size() == 0){
System.out.println("消費者："+ Thread.currentThread().getName()+"，緩沖區(qū)為空,消費者進入waiting...");
// 消費者進入等待狀態(tài)
consumerCondition.await();
}
// 從頭部獲取數(shù)據(jù)，并移除元素
Integer val = list.removeFirst();
System.out.println("消費者："+ Thread.currentThread().getName()+"，value：" + val);
// 喚醒所有生產(chǎn)者處于wait()的線程
producerCondition.signalAll();
} catch (Exception e) {
e.printStackTrace();
} finally {
if(flag){
lock.unlock();
}
}
}
}/**
* 生產(chǎn)者
*/
public class Producer extends Thread{
private ContainerDemo container;
private Integer value;
public Producer(ContainerDemo container, Integer value) {
this.container = container;
this.value = value;
}
@Override
public void run() {
container.add(value);
}
}/**
* 消費者
*/
public class Consumer extends Thread{
private ContainerDemo container;
public Consumer(ContainerDemo container) {
this.container = container;
}
@Override
public void run() {
container.get();
}
}/**
* 測試類
*/
public class MyThreadTest {
public static void main(String[] args) {
ContainerDemo container = new ContainerDemo();
List<Thread> threadList = new ArrayList<>();
// 初始化6個生產(chǎn)者線程
for (int i = 0; i < 6; i++) {
threadList.add(new Producer(container, i));
}
// 初始化6個消費者線程
for (int i = 0; i < 6; i++) {
threadList.add(new Consumer(container));
}
// 啟動線程
for (Thread thread : threadList) {
thread.start();
}
}
}運行結果如下：
生產(chǎn)者：Thread-0，add：0
生產(chǎn)者：Thread-1，add：1
生產(chǎn)者：Thread-2，add：2
生產(chǎn)者：Thread-3，緩沖區(qū)已滿,生產(chǎn)者進入waiting...
生產(chǎn)者：Thread-4，緩沖區(qū)已滿,生產(chǎn)者進入waiting...
生產(chǎn)者：Thread-5，緩沖區(qū)已滿,生產(chǎn)者進入waiting...
消費者：Thread-6，value：0
消費者：Thread-7，value：1
生產(chǎn)者：Thread-3，add：3
生產(chǎn)者：Thread-4，add：4
生產(chǎn)者：Thread-5，add：5
消費者：Thread-8，value：2
消費者：Thread-9，value：3
消費者：Thread-10，value：4
消費者：Thread-11，value：5通過ReentrantLock定義兩個Condition，一個表示生產(chǎn)者的Condition，一個表示消費者的Condition，喚醒的時候調(diào)用對應的signalAll()方法就可以解決假死現(xiàn)象。
03、小結
最后我們來總結一下，對于生產(chǎn)者和消費者模型，通過合理的編程實現(xiàn)，可以充分充分發(fā)揮 cpu 多線程的特性，顯著的提升系統(tǒng)處理數(shù)據(jù)的效率。
對于生產(chǎn)者和消費者模型中的假死現(xiàn)象，可以使用ReentrantLock定義兩個Condition，進行交叉喚醒，以解決假死問題。
責任編輯：武曉燕
來源：
潘志的研發(fā)筆記
