生產者/消費者問題的多種Java實現方式
生產者消費者問題是研究多線程程序時繞不開的經典問題之一，它描述是有一塊緩衝區做爲倉庫，生產者能夠將產品放入倉庫，消費者則能夠從倉庫中取走產品。解決生產者/消費者問題的方法可分爲兩類：（1）採用某種機制保護生產者和消費者之間的同步；（2）在生產者和消費者之間創建一個管道。第一種方式有較高的效率，而且易於實現，代碼的可控制性較好，屬於經常使用的模式。第二種管道緩衝區不易控制，被傳輸數據對象不易於封裝等，實用性不強。所以本文只介紹同步機制實現的生產者/消費者問題。html
同步問題核心在於：如何保證同一資源被多個線程併發訪問時的完整性。經常使用的同步方法是採用信號或加鎖機制，保證資源在任意時刻至多被一個線程訪問。Java語言在多線程編程上實現了徹底對象化，提供了對同步機制的良好支持。在Java中一共有四種方法支持同步，其中前三個是同步方法，一個是管道方法。java
（1）wait() / notify()方法編程
（2）await() / signal()方法設計模式
（3）BlockingQueue阻塞隊列方法安全
（4）PipedInputStream / PipedOutputStream多線程
本文只介紹最經常使用的前三種，第四種暫不作討論，有興趣的讀者能夠本身去網上找答案。併發
1、wait() / notify()方法函數
wait() / nofity()方法是基類Object的兩個方法，也就意味着全部Java類都會擁有這兩個方法，這樣，咱們就能夠爲任何對象實現同步機制。測試
wait()方法：當緩衝區已滿/空時，生產者/消費者線程中止本身的執行，放棄鎖，使本身處於等等狀態，讓其餘線程執行。ui
notify()方法：當生產者/消費者向緩衝區放入/取出一個產品時，向其餘等待的線程發出可執行的通知，同時放棄鎖，使本身處於等待狀態。
光看文字可能不太好理解，咱來段代碼就明白了：
import java.util.LinkedList;
/**
* 倉庫類Storage實現緩衝區
*
* Email:530025983@qq.com
*
* @author MONKEY.D.MENG 2011-03-15
*
*/
public class Storage
{
// 倉庫最大存儲量
private final int MAX_SIZE = 100;
// 倉庫存儲的載體
private LinkedList<Object> list = new LinkedList<Object>();
// 生產num個產品
public void produce(int num)
{
// 同步代碼段
synchronized (list)
{
// 若是倉庫剩餘容量不足
while (list.size() + num > MAX_SIZE)
{
System.out.println("【要生產的產品數量】:" + num + "/t【庫存量】:"
+ list.size() + "/t暫時不能執行生產任務!");
try
{
// 因爲條件不知足，生產阻塞
list.wait();
}
catch (InterruptedException e)
{
e.printStackTrace();
}
}
// 生產條件知足狀況下，生產num個產品
for (int i = 1; i <= num; ++i)
{
list.add(new Object());
}
System.out.println("【已經生產產品數】:" + num + "/t【現倉儲量爲】:" + list.size());
list.notifyAll();
}
}
// 消費num個產品
public void consume(int num)
{
// 同步代碼段
synchronized (list)
{
// 若是倉庫存儲量不足
while (list.size() < num)
{
System.out.println("【要消費的產品數量】:" + num + "/t【庫存量】:"
+ list.size() + "/t暫時不能執行生產任務!");
try
{
// 因爲條件不知足，消費阻塞
list.wait();
}
catch (InterruptedException e)
{
e.printStackTrace();
}
}
// 消費條件知足狀況下，消費num個產品
for (int i = 1; i <= num; ++i)
{
list.remove();
}
System.out.println("【已經消費產品數】:" + num + "/t【現倉儲量爲】:" + 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;
}
}
/**
* 生產者類Producer繼承線程類Thread
*
* Email:530025983@qq.com
*
* @author MONKEY.D.MENG 2011-03-15
*
*/
public class Producer extends Thread
{
// 每次生產的產品數量
private int num;
// 所在放置的倉庫
private Storage storage;
// 構造函數，設置倉庫
public Producer(Storage storage)
{
this.storage = storage;
}
// 線程run函數
public void run()
{
produce(num);
}
// 調用倉庫Storage的生產函數
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;
}
}
/**
* 消費者類Consumer繼承線程類Thread
*
* Email:530025983@qq.com
*
* @author MONKEY.D.MENG 2011-03-15
*
*/
public class Consumer extends Thread
{
// 每次消費的產品數量
private int num;
// 所在放置的倉庫
private Storage storage;
// 構造函數，設置倉庫
public Consumer(Storage storage)
{
this.storage = storage;
}
// 線程run函數
public void run()
{
consume(num);
}
// 調用倉庫Storage的生產函數
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;
}
}
/**
* 測試類Test
*
* Email:530025983@qq.com
*
* @author MONKEY.D.MENG 2011-03-15
*
*/
public class Test
{
public static void main(String[] args)
{
// 倉庫對象
Storage storage = new Storage();
// 生產者對象
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);
// 設置生產者產品生產數量
p1.setNum(10);
p2.setNum(10);
p3.setNum(10);
p4.setNum(10);
p5.setNum(10);
p6.setNum(10);
p7.setNum(80);
// 設置消費者產品消費數量
c1.setNum(50);
c2.setNum(20);
c3.setNum(30);
// 線程開始執行
c1.start();
c2.start();
c3.start();
p1.start();
p2.start();
p3.start();
p4.start();
p5.start();
p6.start();
p7.start();
}
}
【要消費的產品數量】:50 【庫存量】:0 暫時不能執行生產任務!
【要消費的產品數量】:30 【庫存量】:0 暫時不能執行生產任務!
【要消費的產品數量】:20 【庫存量】:0 暫時不能執行生產任務!
【已經生產產品數】:10 【現倉儲量爲】:10
【要消費的產品數量】:20 【庫存量】:10 暫時不能執行生產任務!
【要消費的產品數量】:30 【庫存量】:10 暫時不能執行生產任務!
【要消費的產品數量】:50 【庫存量】:10 暫時不能執行生產任務!
【已經生產產品數】:10 【現倉儲量爲】:20
【要消費的產品數量】:50 【庫存量】:20 暫時不能執行生產任務!
【要消費的產品數量】:30 【庫存量】:20 暫時不能執行生產任務!
【已經消費產品數】:20 【現倉儲量爲】:0
【已經生產產品數】:10 【現倉儲量爲】:10
【已經生產產品數】:10 【現倉儲量爲】:20
【已經生產產品數】:80 【現倉儲量爲】:100
【要生產的產品數量】:10 【庫存量】:100 暫時不能執行生產任務!
【已經消費產品數】:30 【現倉儲量爲】:70
【已經消費產品數】:50 【現倉儲量爲】:20
【已經生產產品數】:10 【現倉儲量爲】:30
【已經生產產品數】:10 【現倉儲量爲】:40
看完上述代碼，對wait() / notify()方法實現的同步有了瞭解。你可能會對Storage類中爲何要定義public void produce(int num);和public void consume(int num);方法感到不解，爲何不直接在生產者類Producer和消費者類Consumer中實現這兩個方法，卻要調用Storage類中的實現呢？淡定，後文會有解釋。咱們先往下走。
2、await() / signal()方法
在JDK5.0以後，Java提供了更加健壯的線程處理機制，包括同步、鎖定、線程池等，它們能夠實現更細粒度的線程控制。await()和signal()就是其中用來作同步的兩種方法，它們的功能基本上和wait() / nofity()相同，徹底能夠取代它們，可是它們和新引入的鎖定機制Lock直接掛鉤，具備更大的靈活性。經過在Lock對象上調用newCondition()方法，將條件變量和一個鎖對象進行綁定，進而控制併發程序訪問競爭資源的安全。下面來看代碼：
import java.util.LinkedList;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* 倉庫類Storage實現緩衝區
*
* Email:530025983@qq.com
*
* @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();
// 生產num個產品
public void produce(int num)
{
// 得到鎖
lock.lock();
// 若是倉庫剩餘容量不足
while (list.size() + num > MAX_SIZE)
{
System.out.println("【要生產的產品數量】:" + num + "/t【庫存量】:" + list.size()
+ "/t暫時不能執行生產任務!");
try
{
// 因爲條件不知足，生產阻塞
full.await();
}
catch (InterruptedException e)
{
e.printStackTrace();
}
}
// 生產條件知足狀況下，生產num個產品
for (int i = 1; i <= num; ++i)
{
list.add(new Object());
}
System.out.println("【已經生產產品數】:" + num + "/t【現倉儲量爲】:" + list.size());
// 喚醒其餘全部線程
full.signalAll();
empty.signalAll();
// 釋放鎖
lock.unlock();
}
// 消費num個產品
public void consume(int num)
{
// 得到鎖
lock.lock();
// 若是倉庫存儲量不足
while (list.size() < num)
{
System.out.println("【要消費的產品數量】:" + num + "/t【庫存量】:" + list.size()
+ "/t暫時不能執行生產任務!");
try
{
// 因爲條件不知足，消費阻塞
empty.await();
}
catch (InterruptedException e)
{
e.printStackTrace();
}
}
// 消費條件知足狀況下，消費num個產品
for (int i = 1; i <= num; ++i)
{
list.remove();
}
System.out.println("【已經消費產品數】:" + num + "/t【現倉儲量爲】:" + 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;
}
}
【要消費的產品數量】:50 【庫存量】:0 暫時不能執行生產任務!
【要消費的產品數量】:30 【庫存量】:0 暫時不能執行生產任務!
【已經生產產品數】:10 【現倉儲量爲】:10
【已經生產產品數】:10 【現倉儲量爲】:20
【要消費的產品數量】:50 【庫存量】:20 暫時不能執行生產任務!
【要消費的產品數量】:30 【庫存量】:20 暫時不能執行生產任務!
【已經生產產品數】:10 【現倉儲量爲】:30
【要消費的產品數量】:50 【庫存量】:30 暫時不能執行生產任務!
【已經消費產品數】:20 【現倉儲量爲】:10
【已經生產產品數】:10 【現倉儲量爲】:20
【要消費的產品數量】:30 【庫存量】:20 暫時不能執行生產任務!
【已經生產產品數】:80 【現倉儲量爲】:100
【要生產的產品數量】:10 【庫存量】:100 暫時不能執行生產任務!
【已經消費產品數】:50 【現倉儲量爲】:50
【已經生產產品數】:10 【現倉儲量爲】:60
【已經消費產品數】:30 【現倉儲量爲】:30
【已經生產產品數】:10 【現倉儲量爲】:40
只須要更新倉庫類Storage的代碼便可，生產者Producer、消費者Consumer、測試類Test的代碼均不須要進行任何更改。這樣咱們就知道爲神馬我要在Storage類中定義public void produce(int num);和public void consume(int num);方法，並在生產者類Producer和消費者類Consumer中調用Storage類中的實現了吧。將可能發生的變化集中到一個類中，不影響原有的構架設計，同時無需修改其餘業務層代碼。無心之中，咱們好像使用了某種設計模式，具體是啥我忘記了，啊哈哈，等我想起來再告訴你們~
3、BlockingQueue阻塞隊列方法
BlockingQueue是JDK5.0的新增內容，它是一個已經在內部實現了同步的隊列，實現方式採用的是咱們第2種await() / signal()方法。它能夠在生成對象時指定容量大小。它用於阻塞操做的是put()和take()方法。
put()方法：相似於咱們上面的生產者線程，容量達到最大時，自動阻塞。
take()方法：相似於咱們上面的消費者線程，容量爲0時，自動阻塞。
關於BlockingQueue的內容網上有不少，你們能夠本身搜，我在這很少介紹。下面直接看代碼，跟以往同樣，咱們只須要更改倉庫類Storage的代碼便可：
import java.util.concurrent.LinkedBlockingQueue;
/**
* 倉庫類Storage實現緩衝區
*
* Email:530025983@qq.com
*
* @author MONKEY.D.MENG 2011-03-15
*
*/
public class Storage
{
// 倉庫最大存儲量
private final int MAX_SIZE = 100;
// 倉庫存儲的載體
private LinkedBlockingQueue<Object> list = new LinkedBlockingQueue<Object>(
100);
// 生產num個產品
public void produce(int num)
{
// 若是倉庫剩餘容量爲0
if (list.size() == MAX_SIZE)
{
System.out.println("【庫存量】:" + MAX_SIZE + "/t暫時不能執行生產任務!");
}
// 生產條件知足狀況下，生產num個產品
for (int i = 1; i <= num; ++i)
{
try
{
// 放入產品，自動阻塞
list.put(new Object());
}
catch (InterruptedException e)
{
e.printStackTrace();
}
System.out.println("【現倉儲量爲】:" + list.size());
}
}
// 消費num個產品
public void consume(int num)
{
// 若是倉庫存儲量不足
if (list.size() == 0)
{
System.out.println("【庫存量】:0/t暫時不能執行生產任務!");
}
// 消費條件知足狀況下，消費num個產品
for (int i = 1; i <= num; ++i)
{
try
{
// 消費產品，自動阻塞
list.take();
}
catch (InterruptedException e)
{
e.printStackTrace();
}
}
System.out.println("【現倉儲量爲】:" + list.size());
}
// set/get方法
public LinkedBlockingQueue<Object> getList()
{
return list;
}
public void setList(LinkedBlockingQueue<Object> list)
{
this.list = list;
}
public int getMAX_SIZE()
{
return MAX_SIZE;
}
}
【庫存量】:0 暫時不能執行生產任務!
【庫存量】:0 暫時不能執行生產任務!
【現倉儲量爲】:1
【現倉儲量爲】:1
【現倉儲量爲】:3
【現倉儲量爲】:4
【現倉儲量爲】:5
【現倉儲量爲】:6
【現倉儲量爲】:7
【現倉儲量爲】:8
【現倉儲量爲】:9
【現倉儲量爲】:10
【現倉儲量爲】:11
【現倉儲量爲】:1
【現倉儲量爲】:2
【現倉儲量爲】:13
【現倉儲量爲】:14
【現倉儲量爲】:17
【現倉儲量爲】:19
【現倉儲量爲】:20
【現倉儲量爲】:21
【現倉儲量爲】:22
【現倉儲量爲】:23
【現倉儲量爲】:24
【現倉儲量爲】:25
【現倉儲量爲】:26
【現倉儲量爲】:12
【現倉儲量爲】:1
【現倉儲量爲】:1
【現倉儲量爲】:2
【現倉儲量爲】:3
【現倉儲量爲】:4
【現倉儲量爲】:5
【現倉儲量爲】:6
【現倉儲量爲】:7
【現倉儲量爲】:27
【現倉儲量爲】:8
【現倉儲量爲】:6
【現倉儲量爲】:18
【現倉儲量爲】:2
【現倉儲量爲】:3
【現倉儲量爲】:4
【現倉儲量爲】:5
【現倉儲量爲】:6
【現倉儲量爲】:7
【現倉儲量爲】:8
【現倉儲量爲】:9
【現倉儲量爲】:10
【現倉儲量爲】:16
【現倉儲量爲】:11
【現倉儲量爲】:12
【現倉儲量爲】:13
【現倉儲量爲】:14
【現倉儲量爲】:15
【現倉儲量爲】:1
【現倉儲量爲】:2
【現倉儲量爲】:3
【現倉儲量爲】:3
【現倉儲量爲】:15
【現倉儲量爲】:1
【現倉儲量爲】:0
【現倉儲量爲】:1
【現倉儲量爲】:1
【現倉儲量爲】:1
【現倉儲量爲】:2
【現倉儲量爲】:3
【現倉儲量爲】:4
【現倉儲量爲】:0
【現倉儲量爲】:1
【現倉儲量爲】:5
【現倉儲量爲】:6
【現倉儲量爲】:7
【現倉儲量爲】:8
【現倉儲量爲】:9
【現倉儲量爲】:10
【現倉儲量爲】:11
【現倉儲量爲】:12
【現倉儲量爲】:13
【現倉儲量爲】:14
【現倉儲量爲】:15
【現倉儲量爲】:16
【現倉儲量爲】:17
【現倉儲量爲】:1
【現倉儲量爲】:1
【現倉儲量爲】:2
【現倉儲量爲】:3
【現倉儲量爲】:4
【現倉儲量爲】:5
【現倉儲量爲】:6
【現倉儲量爲】:3
【現倉儲量爲】:3
【現倉儲量爲】:1
【現倉儲量爲】:2
【現倉儲量爲】:3
【現倉儲量爲】:4
【現倉儲量爲】:5
【現倉儲量爲】:6
【現倉儲量爲】:7
【現倉儲量爲】:8
【現倉儲量爲】:9
【現倉儲量爲】:10
【現倉儲量爲】:11
【現倉儲量爲】:12
【現倉儲量爲】:13
【現倉儲量爲】:14
【現倉儲量爲】:15
【現倉儲量爲】:16
【現倉儲量爲】:17
【現倉儲量爲】:18
【現倉儲量爲】:19
【現倉儲量爲】:6
【現倉儲量爲】:7
【現倉儲量爲】:8
【現倉儲量爲】:9
【現倉儲量爲】:10
【現倉儲量爲】:11
【現倉儲量爲】:12
【現倉儲量爲】:13
【現倉儲量爲】:14
【現倉儲量爲】:15
【現倉儲量爲】:16
【現倉儲量爲】:17
【現倉儲量爲】:18
【現倉儲量爲】:19
【現倉儲量爲】:20
【現倉儲量爲】:21
【現倉儲量爲】:22
【現倉儲量爲】:23
【現倉儲量爲】:24
【現倉儲量爲】:25
【現倉儲量爲】:26
【現倉儲量爲】:27
【現倉儲量爲】:28
【現倉儲量爲】:29
【現倉儲量爲】:30
【現倉儲量爲】:31
【現倉儲量爲】:32
【現倉儲量爲】:33
【現倉儲量爲】:34
【現倉儲量爲】:35
【現倉儲量爲】:36
【現倉儲量爲】:37
【現倉儲量爲】:38
【現倉儲量爲】:39
【現倉儲量爲】:40
固然，你會發現這時對於public void produce(int num);和public void consume(int num);方法業務邏輯上的實現跟前面兩個例子不太同樣，不要緊，這個例子只是爲了說明BlockingQueue阻塞隊列的使用。
有時使用BlockingQueue可能會出現put()和System.out.println()輸出不匹配的狀況，這是因爲它們之間沒有同步形成的。當緩衝區已滿，生產者在put()操做時，put()內部調用了await()方法，放棄了線程的執行，而後消費者線程執行，調用take()方法，take()內部調用了signal()方法，通知生產者線程能夠執行，導致在消費者的println()還沒運行的狀況下生產者的println()先被執行，因此有了輸出不匹配的狀況。
對於BlockingQueue你們能夠放心使用，這可不是它的問題，只是在它和別的對象之間的同步有問題。
推薦閱讀：
相關文章
- 1. 生產者/消費者問題的多種Java實現方式
- 2. 生產者消費者模式的三種實現方式
- 3. 生產者-消費者模式的三種實現方式
- 4. JAVA生產者消費者的實現
- 5. 生產者消費者問題——C++ windows版 多生產者多消費者的隊列實現
- 6. 【Java】生產者消費者模式的三種實現
- 7. Java實現生產者和消費者的三種方法
- 8. 用Java實現生產者消費者的三種方法
- 9. 用 Java 實現生產者和消費者的三種方法
- 10. 生產者消費者的四種實現方式
- 更多相關文章...
- • 現實生活中的 XML - XML 教程
- • Mybatis實現映射器的2種方式 - MyBatis教程
- • ☆基於Java Instrument的Agent實現
- • 適用於PHP初學者的學習線路和建議
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. No provider available from registry 127.0.0.1:2181 for service com.ddbuy.ser 解決方法
- 2. Qt5.7以上調用虛擬鍵盤(支持中文)，以及源碼修改(可拖動，水平縮放)
- 3. 軟件測試面試- 購物車功能測試用例設計
- 4. ElasticSearch（概念篇）：你知道的, 爲了搜索…
- 5. redux理解
- 6. gitee創建第一個項目
- 7. 支持向量機之硬間隔（一步步推導，通俗易懂）
- 8. Mysql 異步複製延遲的原因及解決方案
- 9. 如何在運行SEPM配置嚮導時將不可認的複雜數據庫密碼改爲簡單密碼
- 10. windows系統下tftp服務器使用
歡迎關注本站公眾號,獲取更多信息
