java 中怎么實現(xiàn)生產(chǎn)者消費者

今天就跟大家聊聊有關(guān)java 中怎么實現(xiàn)生產(chǎn)者消費者，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結(jié)了以下內(nèi)容，希望大家根據(jù)這篇文章可以有所收獲。

生產(chǎn)者消費者圖

存儲空間已滿，而生產(chǎn)者占用著它，消費者等著生產(chǎn)者讓出空間從而去除產(chǎn)品，生產(chǎn)者等著消費者消費產(chǎn)品，從而向空間中添加產(chǎn)品?；ハ嗟却瑥亩l(fā)生死鎖。

生產(chǎn)者消費者問題是研究多線程程序時繞不開的經(jīng)典問題之一，它描述是有一塊緩沖區(qū)作為倉庫，生產(chǎn)者可以將產(chǎn)品放入倉庫，消費者則可以從倉庫中取走產(chǎn)品。解決生產(chǎn)者/消費者問題的方法可分為兩類：

（1）采用某種機制保護生產(chǎn)者和消費者之間的同步；

（2）在生產(chǎn)者和消費者之間建立一個管道。

***種方式有較高的效率，并且易于實現(xiàn)，代碼的可控制性較好，屬于常用的模式。第二種管道緩沖區(qū)不易控制，被傳輸數(shù)據(jù)對象不易于封裝等，實用性不強。因此本文只介紹同步機制實現(xiàn)的生產(chǎn)者/消費者問題。

同步問題核心在于：如何保證同一資源被多個線程并發(fā)訪問時的完整性。常用的同步方法是采用信號或加鎖機制， 保證資源在任意時刻至多被一個線程訪問。Java語言在多線程編程上實現(xiàn)了完全對象化，提供了對同步機制的良好支持。在Java中一共有四種方法支持同 步，其中前三個是同步方法，一個是管道方法。

（1）wait() / notify()方法

（2）await() / signal()方法

（3）BlockingQueue阻塞隊列方法

（4）PipedInputStream / PipedOutputStream

本文只介紹最常用的前兩種種，第三、四種暫不做討論，有興趣的讀者可以自己去網(wǎng)上找答案。

一、wait() / notify()方法

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)：

1、倉庫類

import java.util.LinkedList;    /**   * 倉庫類Storage實現(xiàn)緩沖區(qū)   *    * @author zcr   */   public class Storage   {       // 倉庫***存儲量       private final int MAX_SIZE = 100;          // 倉庫存儲的載體       private LinkedList<Object> list = new LinkedList<Object>();          /**      * 生產(chǎn)num個產(chǎn)品      * @param num 生產(chǎn)產(chǎn)品的數(shù)量      */     public void produce(int num)       {           // 同步代碼段           synchronized (list)           {               // 如果倉庫剩余容量不足               while (list.size() + num > MAX_SIZE)               {                   System.out.println("【要生產(chǎn)的產(chǎn)品數(shù)量】:" + num + " \t 【庫存量】:"                           + list.size() + "\t 暫時不能執(zhí)行生產(chǎn)任務(wù)!");                   try                   {                       // 由于條件不滿足，生產(chǎn)阻塞                       list.wait();                   }                   catch (InterruptedException e)                   {                       e.printStackTrace();                   }               }                  // 生產(chǎn)條件滿足情況下，生產(chǎn)num個產(chǎn)品               for (int i = 1; i <= num; ++i)               {                   list.add(new Object());               }                  System.out.println("【已經(jīng)生產(chǎn)產(chǎn)品數(shù)】:" + num + "\t 【現(xiàn)倉儲量為】:" + list.size());                  list.notifyAll();           }       }          /**      * 消費num個產(chǎn)品      * @param num 消費產(chǎn)品數(shù)量      */     public void consume(int num)       {           // 同步代碼段           synchronized (list)           {               // 如果倉庫存儲量不足               while (list.size() < num)               {                   System.out.println("【要消費的產(chǎn)品數(shù)量】:" + num + " \t【庫存量】:"                           + list.size() + " \t 暫時不能執(zhí)行生產(chǎn)任務(wù)!");                   try                   {                       // 由于條件不滿足，消費阻塞                       list.wait();                   }                   catch (InterruptedException e)                   {                       e.printStackTrace();                   }               }                  // 消費條件滿足情況下，消費num個產(chǎn)品               for (int i = 1; i <= num; ++i)               {                   list.remove();               }                  System.out.println("【已經(jīng)消費產(chǎn)品數(shù)】:" + num + " \t 【現(xiàn)倉儲量為】:" + 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;       }   }

2、生產(chǎn)者

/**   * 生產(chǎn)者類Producer繼承線程類Thread   *    *    * @author zcr  *    */   public class Producer extends Thread   {       // 每次生產(chǎn)的產(chǎn)品數(shù)量       private int num;          // 所在放置的倉庫       private Storage storage;          // 構(gòu)造函數(shù)，設(shè)置倉庫       public Producer(Storage storage)       {           this.storage = storage;       }          // 線程run函數(shù)       public void run()       {           produce(num);       }          // 調(diào)用倉庫Storage的生產(chǎn)函數(shù)       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;       }   }

3、消費者

/**   * 消費者類Consumer繼承線程類Thread   *    *    * @author zcr  *    */   public class Consumer extends Thread   {       // 每次消費的產(chǎn)品數(shù)量       private int num;          // 所在放置的倉庫       private Storage storage;          // 構(gòu)造函數(shù)，設(shè)置倉庫       public Consumer(Storage storage)       {           this.storage = storage;       }          // 線程run函數(shù)       public void run()       {           consume(num);       }          // 調(diào)用倉庫Storage的生產(chǎn)函數(shù)       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;       }   }

4、測試類

/**   * 測試類Test   * @author zcr  *    */   public class Test   {       public static void main(String[] args)       {           // 倉庫對象           Storage storage = new Storage();              // 生產(chǎn)者對象           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);              // 設(shè)置生產(chǎn)者產(chǎn)品生產(chǎn)數(shù)量           p1.setNum(10);           p2.setNum(10);           p3.setNum(10);           p4.setNum(10);           p5.setNum(10);           p6.setNum(10);           p7.setNum(80);              // 設(shè)置消費者產(chǎn)品消費數(shù)量           c1.setNum(50);           c2.setNum(20);           c3.setNum(30);              // 線程開始執(zhí)行           c1.start();           c2.start();           c3.start();           p1.start();           p2.start();           p3.start();           p4.start();           p5.start();           p6.start();           p7.start();       }   }

5、結(jié)果：

【要消費的產(chǎn)品數(shù)量】:50     【庫存量】:0      暫時不能執(zhí)行生產(chǎn)任務(wù)! 【要消費的產(chǎn)品數(shù)量】:20     【庫存量】:0      暫時不能執(zhí)行生產(chǎn)任務(wù)! 【已經(jīng)生產(chǎn)產(chǎn)品數(shù)】:10     【現(xiàn)倉儲量為】:10 【要消費的產(chǎn)品數(shù)量】:20     【庫存量】:10      暫時不能執(zhí)行生產(chǎn)任務(wù)! 【要消費的產(chǎn)品數(shù)量】:50     【庫存量】:10      暫時不能執(zhí)行生產(chǎn)任務(wù)! 【已經(jīng)生產(chǎn)產(chǎn)品數(shù)】:10     【現(xiàn)倉儲量為】:20 【已經(jīng)生產(chǎn)產(chǎn)品數(shù)】:10     【現(xiàn)倉儲量為】:30 【要消費的產(chǎn)品數(shù)量】:50     【庫存量】:30      暫時不能執(zhí)行生產(chǎn)任務(wù)! 【已經(jīng)消費產(chǎn)品數(shù)】:20      【現(xiàn)倉儲量為】:10 【已經(jīng)生產(chǎn)產(chǎn)品數(shù)】:10     【現(xiàn)倉儲量為】:20 【已經(jīng)生產(chǎn)產(chǎn)品數(shù)】:10     【現(xiàn)倉儲量為】:30 【要消費的產(chǎn)品數(shù)量】:50     【庫存量】:30      暫時不能執(zhí)行生產(chǎn)任務(wù)! 【已經(jīng)生產(chǎn)產(chǎn)品數(shù)】:10     【現(xiàn)倉儲量為】:40 【已經(jīng)消費產(chǎn)品數(shù)】:30      【現(xiàn)倉儲量為】:10 【要消費的產(chǎn)品數(shù)量】:50     【庫存量】:10      暫時不能執(zhí)行生產(chǎn)任務(wù)! 【已經(jīng)生產(chǎn)產(chǎn)品數(shù)】:80     【現(xiàn)倉儲量為】:90 【已經(jīng)消費產(chǎn)品數(shù)】:50      【現(xiàn)倉儲量為】:40

看完上述代碼，對wait() / notify()方法實現(xiàn)的同步有了了解。你可能會對Storage類中為什么要定義public void produce(int num);和public void consume(int num);方法感到不解，為什么不直接在生產(chǎn)者類Producer和消費者類Consumer中實現(xiàn)這兩個方法，卻要調(diào)用Storage類中的實現(xiàn)呢？淡定，后文會有解釋。我們先往下走。

二、await() / signal()方法

在JDK5.0之后，Java提供了更加健壯的線程處理機制，包括同步、鎖定、線程池等，它們可以實現(xiàn)更細粒度的線程控制。await()和 signal()就是其中用來做同步的兩種方法，它們的功能基本上和wait() / nofity()相同，完全可以取代它們，但是它們和新引入的鎖定機制Lock直接掛鉤，具有更大的靈活性。通過在Lock對象上調(diào)用newCondition()方法，將條件變量和一個鎖對象進行綁定，進而控制并發(fā)程序訪問競爭資源的安全。

下面來看代碼：

只需要更新倉庫類Storage的代碼即可，生產(chǎn)者Producer、消費者Consumer、測試類Test的代碼均不需要進行任何更改。

倉庫類

import java.util.LinkedList; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock;  /**  * 倉庫類Storage實現(xiàn)緩沖區(qū)  *   *   * @author MONKEY.D.MENG 2011-03-15  *   */ public class Storage {     // 倉庫***存儲量     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();      // 生產(chǎn)num個產(chǎn)品     public void produce(int num)     {         // 獲得鎖         lock.lock();          // 如果倉庫剩余容量不足         while (list.size() + num > MAX_SIZE)         {             System.out.println("【要生產(chǎn)的產(chǎn)品數(shù)量】:" + num + "/t【庫存量】:" + list.size()                     + "/t暫時不能執(zhí)行生產(chǎn)任務(wù)!");             try             {                 // 由于條件不滿足，生產(chǎn)阻塞                 full.await();             }             catch (InterruptedException e)             {                 e.printStackTrace();             }         }          // 生產(chǎn)條件滿足情況下，生產(chǎn)num個產(chǎn)品         for (int i = 1; i <= num; ++i)         {             list.add(new Object());         }          System.out.println("【已經(jīng)生產(chǎn)產(chǎn)品數(shù)】:" + num + "/t【現(xiàn)倉儲量為】:" + list.size());          // 喚醒其他所有線程         full.signalAll();         empty.signalAll();          // 釋放鎖         lock.unlock();     }      // 消費num個產(chǎn)品     public void consume(int num)     {         // 獲得鎖         lock.lock();          // 如果倉庫存儲量不足         while (list.size() < num)         {             System.out.println("【要消費的產(chǎn)品數(shù)量】:" + num + "/t【庫存量】:" + list.size()                     + "/t暫時不能執(zhí)行生產(chǎn)任務(wù)!");             try             {                 // 由于條件不滿足，消費阻塞                 empty.await();             }             catch (InterruptedException e)             {                 e.printStackTrace();             }         }          // 消費條件滿足情況下，消費num個產(chǎn)品         for (int i = 1; i <= num; ++i)         {             list.remove();         }          System.out.println("【已經(jīng)消費產(chǎn)品數(shù)】:" + num + "/t【現(xiàn)倉儲量為】:" + 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;     } }

結(jié)果：

【要消費的產(chǎn)品數(shù)量】:50/t【庫存量】:0/t暫時不能執(zhí)行生產(chǎn)任務(wù)! 
【已經(jīng)生產(chǎn)產(chǎn)品數(shù)】:10/t【現(xiàn)倉儲量為】:10 
【已經(jīng)生產(chǎn)產(chǎn)品數(shù)】:10/t【現(xiàn)倉儲量為】:20 
【要消費的產(chǎn)品數(shù)量】:30/t【庫存量】:20/t暫時不能執(zhí)行生產(chǎn)任務(wù)! 
【已經(jīng)消費產(chǎn)品數(shù)】:20/t【現(xiàn)倉儲量為】:0 
【已經(jīng)生產(chǎn)產(chǎn)品數(shù)】:10/t【現(xiàn)倉儲量為】:10 
【已經(jīng)生產(chǎn)產(chǎn)品數(shù)】:10/t【現(xiàn)倉儲量為】:20 
【要消費的產(chǎn)品數(shù)量】:50/t【庫存量】:20/t暫時不能執(zhí)行生產(chǎn)任務(wù)! 
【已經(jīng)生產(chǎn)產(chǎn)品數(shù)】:10/t【現(xiàn)倉儲量為】:30 
【已經(jīng)生產(chǎn)產(chǎn)品數(shù)】:10/t【現(xiàn)倉儲量為】:40 
【要生產(chǎn)的產(chǎn)品數(shù)量】:80/t【庫存量】:40/t暫時不能執(zhí)行生產(chǎn)任務(wù)! 
【已經(jīng)消費產(chǎn)品數(shù)】:30/t【現(xiàn)倉儲量為】:10 
【要消費的產(chǎn)品數(shù)量】:50/t【庫存量】:10/t暫時不能執(zhí)行生產(chǎn)任務(wù)! 
【已經(jīng)生產(chǎn)產(chǎn)品數(shù)】:80/t【現(xiàn)倉儲量為】:90 
【已經(jīng)消費產(chǎn)品數(shù)】:50/t【現(xiàn)倉儲量為】:40 

這樣我們就知道為神馬我要在Storage類中定義public void produce(int num);和public void consume(int num);方法，并在生產(chǎn)者類Producer和消費者類Consumer中調(diào)用Storage類中的實現(xiàn)了吧。將可能發(fā)生的變化集中到一個類中，不影響原有的構(gòu)架設(shè)計，同時無需修改其他業(yè)務(wù)層代碼。

看完上述內(nèi)容，你們對java 中怎么實現(xiàn)生產(chǎn)者消費者有進一步的了解嗎？如果還想了解更多知識或者相關(guān)內(nèi)容，請關(guān)注億速云行業(yè)資訊頻道，感謝大家的支持。