Java多線程并發(fā)生產(chǎn)者消費者設(shè)計模式實例解析
一、兩個線程一個生產(chǎn)者一個消費者
需求情景
兩個線程，一個負責(zé)生產(chǎn)，一個負責(zé)消費，生產(chǎn)者生產(chǎn)一個，消費者消費一個。
涉及問題
- 同步問題：如何保證同一資源被多個線程并發(fā)訪問時的完整性。常用的同步方法是采用標(biāo)記或加鎖機制。
- wait() / nofity() 方法是基類Object的兩個方法，也就意味著所有Java類都會擁有這兩個方法，這樣，我們就可以為任何對象實現(xiàn)同步機制。
- wait()方法：當(dāng)緩沖區(qū)已滿/空時，生產(chǎn)者/消費者線程停止自己的執(zhí)行，放棄鎖，使自己處于等待狀態(tài)，讓其他線程執(zhí)行。
- notify()方法：當(dāng)生產(chǎn)者/消費者向緩沖區(qū)放入/取出一個產(chǎn)品時，向其他等待的線程發(fā)出可執(zhí)行的通知，同時放棄鎖，使自己處于等待狀態(tài)。
代碼實現(xiàn)（共三個類和一個main方法的測試類）
Resource.java
package com.demo.ProducerConsumer;
/**
* 資源
* @author lixiaoxi
*
*/
public class Resource {
/*資源序號*/
private int number = 0;
/*資源標(biāo)記*/
private boolean flag = false;
/**
* 生產(chǎn)資源
*/
public synchronized void create() {
if (flag) {//先判斷標(biāo)記是否已經(jīng)生產(chǎn)了，如果已經(jīng)生產(chǎn)，等待消費；
try {
wait();//讓生產(chǎn)線程等待
} catch (InterruptedException e) {
e.printStackTrace();
}
}
number++;//生產(chǎn)一個
System.out.println(Thread.currentThread().getName() + "生產(chǎn)者------------" + number);
flag = true;//將資源標(biāo)記為已經(jīng)生產(chǎn)
notify();//喚醒在等待操作資源的線程（隊列）
}
/**
* 消費資源
*/
public synchronized void destroy() {
if (!flag) {
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println(Thread.currentThread().getName() + "消費者****" + number);
flag = false;
notify();
}
}
Producer.java
package com.demo.ProducerConsumer;
/**
* 生產(chǎn)者
* @author lixiaoxi
*
*/
public class Producer implements Runnable{
private Resource resource;
public Producer(Resource resource) {
this.resource = resource;
}
@Override
public void run() {
while (true) {
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
resource.create();
}
}
}
Consumer.java
package com.demo.ProducerConsumer;
/**
* 消費者
* @author lixiaoxi
*
*/
public class Consumer implements Runnable{
private Resource resource;
public Consumer(Resource resource) {
this.resource = resource;
}
@Override
public void run() {
while (true) {
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
resource.destroy();
}
}
}
ProducerConsumerTest.java
package com.demo.ProducerConsumer;
public class ProducerConsumerTest {
public static void main(String args[]) {
Resource resource = new Resource();
new Thread(new Producer(resource)).start();//生產(chǎn)者線程
new Thread(new Consumer(resource)).start();//消費者線程
}
}
打印結(jié)果：
以上打印結(jié)果可以看出沒有任何問題。
二、多個線程，多個生產(chǎn)者和多個消費者的問題
需求情景
四個線程，兩個個負責(zé)生產(chǎn)，兩個個負責(zé)消費，生產(chǎn)者生產(chǎn)一個，消費者消費一個。
涉及問題
notifyAll()方法：當(dāng)生產(chǎn)者/消費者向緩沖區(qū)放入/取出一個產(chǎn)品時，向其他等待的所有線程發(fā)出可執(zhí)行的通知，同時放棄鎖，使自己處于等待狀態(tài)。
再次測試代碼
ProducerConsumerTest.java
package com.demo.ProducerConsumer;
public class ProducerConsumerTest {
public static void main(String args[]) {
Resource resource = new Resource();
new Thread(new Producer(resource)).start();//生產(chǎn)者線程
new Thread(new Producer(resource)).start();//生產(chǎn)者線程
new Thread(new Consumer(resource)).start();//消費者線程
new Thread(new Consumer(resource)).start();//消費者線程
}
}
運行結(jié)果：
通過以上打印結(jié)果發(fā)現(xiàn)問題
147生產(chǎn)了一次，消費了兩次。169生產(chǎn)了，而沒有消費。
原因分析
當(dāng)兩個線程同時操作生產(chǎn)者生產(chǎn)或者消費者消費時，如果有生產(chǎn)者或消費者的兩個線程都wait()時，再次notify()，由于其中一個線程已經(jīng)改變了標(biāo)記而另外一個線程再次往下直接執(zhí)行的時候沒有判斷標(biāo)記而導(dǎo)致的。if判斷標(biāo)記，只有一次，會導(dǎo)致不該運行的線程運行了。出現(xiàn)了數(shù)據(jù)錯誤的情況。
解決方案
while判斷標(biāo)記，解決了線程獲取執(zhí)行權(quán)后，是否要運行！也就是每次wait()后再notify()時先再次判斷標(biāo)記。
代碼改進（Resource中的 if -> while）
Resource.java
package com.demo.ProducerConsumer;
/**
* 資源
* @author lixiaoxi
*
*/
public class Resource {
/*資源序號*/
private int number = 0;
/*資源標(biāo)記*/
private boolean flag = false;
/**
* 生產(chǎn)資源
*/
public synchronized void create() {
while (flag) {//先判斷標(biāo)記是否已經(jīng)生產(chǎn)了，如果已經(jīng)生產(chǎn)，等待消費；
try {
wait();//讓生產(chǎn)線程等待
} catch (InterruptedException e) {
e.printStackTrace();
}
}
number++;//生產(chǎn)一個
System.out.println(Thread.currentThread().getName() + "生產(chǎn)者------------" + number);
flag = true;//將資源標(biāo)記為已經(jīng)生產(chǎn)
notify();//喚醒在等待操作資源的線程（隊列）
}
/**
* 消費資源
*/
public synchronized void destroy() {
while (!flag) {
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println(Thread.currentThread().getName() + "消費者****" + number);
flag = false;
notify();
}
}
運行結(jié)果：
再次發(fā)現(xiàn)問題
打印到某個值比如生產(chǎn)完187，程序運行卡死了，好像鎖死了一樣。
原因分析
notify：只能喚醒一個線程，如果本方喚醒了本方，沒有意義。而且while判斷標(biāo)記+notify會導(dǎo)致”死鎖”。
解決方案
notifyAll解決了本方線程一定會喚醒對方線程的問題。
最后代碼改進（Resource中的 notify() -> notifyAll()）
Resource.java
package com.demo.ProducerConsumer;
/**
* 資源
* @author lixiaoxi
*
*/
public class Resource {
/*資源序號*/
private int number = 0;
/*資源標(biāo)記*/
private boolean flag = false;
/**
* 生產(chǎn)資源
*/
public synchronized void create() {
while (flag) {//先判斷標(biāo)記是否已經(jīng)生產(chǎn)了，如果已經(jīng)生產(chǎn)，等待消費；
try {
wait();//讓生產(chǎn)線程等待
} catch (InterruptedException e) {
e.printStackTrace();
}
}
number++;//生產(chǎn)一個
System.out.println(Thread.currentThread().getName() + "生產(chǎn)者------------" + number);
flag = true;//將資源標(biāo)記為已經(jīng)生產(chǎn)
notifyAll();//喚醒在等待操作資源的線程（隊列）
}
/**
* 消費資源
*/
public synchronized void destroy() {
while (!flag) {
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println(Thread.currentThread().getName() + "消費者****" + number);
flag = false;
notifyAll();
}
}
運行結(jié)果：
以上就大功告成了，沒有任何問題。
再來梳理一下整個流程。按照示例，生產(chǎn)者消費者交替運行，每次生產(chǎn)后都有對應(yīng)的消費者，測試類創(chuàng)建實例，如果是生產(chǎn)者先運行，進入run()方法，進入create()方法，flag默認為false，number+1，生產(chǎn)者生產(chǎn)一個產(chǎn)品，flag置為true，同時調(diào)用notifyAll()方法，喚醒所有正在等待的線程，接下來如果還是生產(chǎn)者運行呢？這是flag為true，進入while循環(huán)，執(zhí)行wait()方法，接下來如果是消費者運行的話，調(diào)用destroy()方法，這時flag為true，消費者購買了一次產(chǎn)品，隨即將flag置為false，并喚醒所有正在等待的線程。這就是一次完整的多生產(chǎn)者對應(yīng)多消費者的問題。
三、使用Lock和Condition來解決生產(chǎn)者消費者問題
上面的代碼有一個問題，就是我們?yōu)榱吮苊馑械木€程都處于等待的狀態(tài)，使用了notifyAll方法來喚醒所有的線程，即notifyAll喚醒的是自己方和對方線程。如果我需要只是喚醒對方的線程，比如：生產(chǎn)者只能喚醒消費者的線程，消費者只能喚醒生產(chǎn)者的線程。
在jdk1.5當(dāng)中為我們提供了多線程的升級解決方案：
1. 將同步synchronized替換成了Lock操作。
2. 將Object中的wait，notify，notifyAll方法替換成了Condition對象。
3. 可以只喚醒對方的線程。
完整代碼：
Resource1.java
package com.demo.ProducerConsumer;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* 資源
* @author lixiaoxi
*
*/
public class Resource1 {
/*資源序號*/
private int number = 0;
/*資源標(biāo)記*/
private boolean flag = false;
private Lock lock = new ReentrantLock();
//使用lock建立生產(chǎn)者的condition對象
private Condition condition_pro = lock.newCondition();
//使用lock建立消費者的condition對象
private Condition condition_con = lock.newCondition();
/**
* 生產(chǎn)資源
*/
public void create() throws InterruptedException {
try{
lock.lock();
//先判斷標(biāo)記是否已經(jīng)生產(chǎn)了，如果已經(jīng)生產(chǎn)，等待消費
while(flag){
//生產(chǎn)者等待
condition_pro.await();
}
//生產(chǎn)一個
number++;
System.out.println(Thread.currentThread().getName() + "生產(chǎn)者------------" + number);
//將資源標(biāo)記為已經(jīng)生產(chǎn)
flag = true;
//生產(chǎn)者生產(chǎn)完畢后，喚醒消費者的線程（注意這里不是signalAll)
condition_con.signal();
}finally{
lock.unlock();
}
}
/**
* 消費資源
*/
public void destroy() throws InterruptedException{
try{
lock.lock();
//先判斷標(biāo)記是否已經(jīng)消費了，如果已經(jīng)消費，等待生產(chǎn)
while(!flag){
//消費者等待
condition_con.await();
}
System.out.println(Thread.currentThread().getName() + "消費者****" + number);
//將資源標(biāo)記為已經(jīng)消費
flag = false;
//消費者消費完畢后，喚醒生產(chǎn)者的線程
condition_pro.signal();
}finally{
lock.unlock();
}
}
}
Producer1.java
package com.demo.ProducerConsumer;
/**
* 生產(chǎn)者
* @author lixiaoxi
*
*/
public class Producer1 implements Runnable{
private Resource1 resource;
public Producer1(Resource1 resource) {
this.resource = resource;
}
@Override
public void run() {
while (true) {
try {
Thread.sleep(10);
resource.create();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
Consumer1.java
package com.demo.ProducerConsumer;
/**
* 消費者
* @author lixiaoxi
*
*/
public class Consumer1 implements Runnable{
private Resource1 resource;
public Consumer1(Resource1 resource) {
this.resource = resource;
}
@Override
public void run() {
while (true) {
try {
Thread.sleep(10);
resource.destroy();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
ProducerConsumerTest1.java
package com.demo.ProducerConsumer;
public class ProducerConsumerTest1 {
public static void main(String args[]) {
Resource1 resource = new Resource1();
new Thread(new Producer1(resource)).start();//生產(chǎn)者線程
new Thread(new Producer1(resource)).start();//生產(chǎn)者線程
new Thread(new Consumer1(resource)).start();//消費者線程
new Thread(new Consumer1(resource)).start();//消費者線程
}
}
運行結(jié)果：
四、總結(jié)
1、如果生產(chǎn)者、消費者都是1個，那么flag標(biāo)記可以用if判斷。這里有多個，必須用while判斷。
2、在while判斷的同時，notify函數(shù)可能喚醒本類線程(如一個消費者喚醒另一個消費者)，這會導(dǎo)致所有消費者忙等待，程序無法繼續(xù)往下執(zhí)行。使用notifyAll函數(shù)代替notify可以解決這個問題，notifyAll可以保證非本類線程被喚醒(消費者線程能喚醒生產(chǎn)者線程，反之也可以)，解決了忙等待問題。
小心假死
生產(chǎn)者/消費者模型最終達到的目的是平衡生產(chǎn)者和消費者的處理能力，達到這個目的的過程中，并不要求只有一個生產(chǎn)者和一個消費者。可以多個生產(chǎn)者對應(yīng)多個消費者，可以一個生產(chǎn)者對應(yīng)一個消費者，可以多個生產(chǎn)者對應(yīng)一個消費者。
假死就發(fā)生在上面三種場景下。假死指的是全部線程都進入了WAITING狀態(tài)，那么程序就不再執(zhí)行任何業(yè)務(wù)功能了，整個項目呈現(xiàn)停滯狀態(tài)。
比方說有生產(chǎn)者A和生產(chǎn)者B，緩沖區(qū)由于空了，消費者處于WAITING。生產(chǎn)者B處于WAITING，生產(chǎn)者A被消費者通知生產(chǎn)，生產(chǎn)者A生產(chǎn)出來的產(chǎn)品本應(yīng)該通知消費者，結(jié)果通知了生產(chǎn)者B，生產(chǎn)者B被喚醒，發(fā)現(xiàn)緩沖區(qū)滿了，于是繼續(xù)WAITING。至此，兩個生產(chǎn)者線程處于WAITING，消費者處于WAITING，系統(tǒng)假死。
上面的分析可以看出，假死出現(xiàn)的原因是因為notify的是同類，所以非單生產(chǎn)者/單消費者的場景，可以采取兩種方法解決這個問題：
（1）synchronized用notifyAll()喚醒所有線程、ReentrantLock用signalAll()喚醒所有線程。
（2）用ReentrantLock定義兩個Condition，一個表示生產(chǎn)者的Condition，一個表示消費者的Condition，喚醒的時候調(diào)用相應(yīng)的Condition的signal()方法就可以了。
以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。
相關(guān)文章
這篇文章主要介紹了IDEA的下載和使用安裝，本文通過圖文并茂的形式給大家介紹的非常詳細，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友可以參考下2020-08-08
Springboot項目通過redis實現(xiàn)接口的冪等性
這篇文章主要為大家介紹了Springboot項目通過redis實現(xiàn)接口的冪等性,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步,早日升職加薪2023-12-12
Spring的Aware接口實現(xiàn)及執(zhí)行順序詳解
這篇文章主要為大家介紹了Spring的Aware接口實現(xiàn)及執(zhí)行順序詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2022-12-12
java中關(guān)于轉(zhuǎn)義字符的一個bug
本文主要介紹了java中關(guān)于轉(zhuǎn)義字符的一個bug。具有很好的參考價值，下面跟著小編一起來看下吧2017-02-02
在配合配置中心修改配置讓應(yīng)用自動刷新配置時，我們要在需要感知配置變化的bean上面加上@RereshScope。如果我們不加上這注解，那么有可能無法完成配置自動刷新。本文就來和大家講講@RereshScope刷新的原理，需要的可以參考一下2022-12-12
Fluent Mybatis學(xué)習(xí)之Update語法實踐
Fluent MyBatis是一個MyBatis的增強工具，沒有對mybatis做任何修改。本篇文章將詳細介紹對Fluent Mybatis中的update語法進行驗證。代碼具有一定價值，感興趣的小伙伴可以學(xué)習(xí)一下2021-11-11
本文主要介紹了java中創(chuàng)建線程的幾種方式。具有很好的參考價值，下面跟著小編一起來看下吧2017-02-02
Java基礎(chǔ)之并發(fā)相關(guān)知識總結(jié)
隨著摩爾定律逐步失效,cpu單核性能達到瓶頸,并發(fā)逐漸逐漸得到廣泛應(yīng)用,因而學(xué)習(xí)了解以及使用并發(fā)就顯得十分重要,但并發(fā)相關(guān)的知識比較瑣碎,不易系統(tǒng)學(xué)習(xí),因而本篇文章參照王寶令老師《Java并發(fā)編程》來勾勒出一張“并發(fā)全景圖”,需要的朋友可以參考下2021-05-05
Java用單向環(huán)形鏈表來解決約瑟夫環(huán)Josepfu問題
如果把單鏈表的最后一個節(jié)點的指針指向鏈表頭部，而不是指向NULL，那么就構(gòu)成了一個單向循環(huán)鏈表，通俗講就是把尾節(jié)點的下一跳指向頭結(jié)點2021-10-10
在Java Spring框架中使用的設(shè)計模式有哪些
面試中常會被問道Spring框架使用了哪些設(shè)計模式？關(guān)于這個問題本文給大家介紹的非常詳細，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值，需要的朋友可以參考下2021-09-09