一、兩個線程一個生產(chǎn)者一個消費者

需求情景

兩個線程，一個負責生產(chǎn)，一個負責消費，生產(chǎn)者生產(chǎn)一個，消費者消費一個。

涉及問題

同步問題：如何保證同一資源被多個線程并發(fā)訪問時的完整性。常用的同步方法是采用標記或加鎖機制。 wait() / nofity() 方法是基類Object的兩個方法，也就意味著所有Java類都會擁有這兩個方法，這樣，我們就可以為任何對象實現(xiàn)同步機制。 wait()方法：當緩沖區(qū)已滿/空時，生產(chǎn)者/消費者線程停止自己的執(zhí)行，放棄鎖，使自己處于等待狀態(tài)，讓其他線程執(zhí)行。 notify()方法：當生產(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; /*資源標記*/ private boolean flag = false; /** * 生產(chǎn)資源 */ public synchronized void create() { if (flag) {//先判斷標記是否已經(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;//將資源標記為已經(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)者和多個消費者的問題

需求情景

四個線程，兩個個負責生產(chǎn)，兩個個負責消費，生產(chǎn)者生產(chǎn)一個，消費者消費一個。

涉及問題

notifyAll()方法：當生產(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)了，而沒有消費。

原因分析

當兩個線程同時操作生產(chǎn)者生產(chǎn)或者消費者消費時，如果有生產(chǎn)者或消費者的兩個線程都wait()時，再次notify()，由于其中一個線程已經(jīng)改變了標記而另外一個線程再次往下直接執(zhí)行的時候沒有判斷標記而導(dǎo)致的。if判斷標記，只有一次，會導(dǎo)致不該運行的線程運行了。出現(xiàn)了數(shù)據(jù)錯誤的情況。

解決方案

while判斷標記，解決了線程獲取執(zhí)行權(quán)后，是否要運行！也就是每次wait()后再notify()時先再次判斷標記。

代碼改進（Resource中的 if -> while）

Resource.java

package com.demo.ProducerConsumer;/** * 資源 * @author lixiaoxi * */public class Resource { /*資源序號*/ private int number = 0; /*資源標記*/ private boolean flag = false; /** * 生產(chǎn)資源 */ public synchronized void create() { while (flag) {//先判斷標記是否已經(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;//將資源標記為已經(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判斷標記+notify會導(dǎo)致”死鎖”。

解決方案

notifyAll解決了本方線程一定會喚醒對方線程的問題。

最后代碼改進（Resource中的 notify() -> notifyAll()）

Resource.java

package com.demo.ProducerConsumer;/** * 資源 * @author lixiaoxi * */public class Resource { /*資源序號*/ private int number = 0; /*資源標記*/ private boolean flag = false; /** * 生產(chǎn)資源 */ public synchronized void create() { while (flag) {//先判斷標記是否已經(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;//將資源標記為已經(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當中為我們提供了多線程的升級解決方案：

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; /*資源標記*/ 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(); //先判斷標記是否已經(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); //將資源標記為已經(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(); //先判斷標記是否已經(jīng)消費了，如果已經(jīng)消費，等待生產(chǎn) while(!flag){//消費者等待condition_con.await(); } System.out.println(Thread.currentThread().getName() + '消費者****' + number); //將資源標記為已經(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標記可以用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é)習有所幫助，也希望大家多多支持好吧啦網(wǎng)。