java 生産者消費者問題

引言

生産者和消費者問題是線程模型中的經典問題：生産者和消費者在同一時間段内共用同一個存儲空間，如下圖所示，生産者向空間裡存放資料，而消費者取用資料，如果不加以協調可能會出現以下情況：

生産者消費者圖

存儲空間已滿，而生産者占用着它，消費者等着生産者讓出空間進而去除産品，生産者等着消費者消費産品，進而向空間中添加産品。互相等待，進而發生死鎖。

生産者消費者問題是研究多線程程式時繞不開的經典問題之一，它描述是有一塊緩沖區作為倉庫，生産者可以将産品放入倉庫，消費者則可以從倉庫中取走産品。解決生産者/消費者問題的方法可分為兩類：

（1）采用某種機制保護生産者和消費者之間的同步；

（2）在生産者和消費者之間建立一個管道。

第一種方式有較高的效率，并且易于實作，代碼的可控制性較好，屬于常用的模式。第二種管道緩沖區不易控制，被傳輸資料對象不易于封裝等，實用性不強。是以本文隻介紹同步機制實作的生産者/消費者問題。

同步問題核心在于：如何保證同一資源被多個線程并發通路時的完整性。常用的同步方法是采用信号或加鎖機制，

保證資源在任意時刻至多被一個線程通路。java語言在多線程程式設計上實作了完全對象化，提供了對同步機制的良好支援。在java中一共有四種方法支援同

步，其中前三個是同步方法，一個是管道方法。

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

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

（3）blockingqueue阻塞隊列方法

（4）pipedinputstream / pipedoutputstream

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

一、wait() / notify()方法

wait() / nofity()方法是基類object的兩個方法，也就意味着所有java類都會擁有這兩個方法，這樣，我們就可以為任何對象實作同步機制。

wait()方法：當緩沖區已滿/空時，生産者/消費者線程停止自己的執行，放棄鎖，使自己處于等等狀态，讓其他線程執行。

notify()方法：當生産者/消費者向緩沖區放入/取出一個産品時，向其他等待的線程發出可執行的通知，同時放棄鎖，使自己處于等待狀态。

代碼實作：

1、倉庫類

<a>import java.util.linkedlist; </a>

<a> * 倉庫類storage實作緩沖區 </a>

<a> * @author zcr </a>

<a>public class storage </a>

<a> // 倉庫最大存儲量 </a>

<a> private final int max_size = 100; </a>

<a> // 倉庫存儲的載體 </a>

<a> private linkedlist<object> list = new linkedlist<object>(); </a>

<a> * 生産num個産品 </a>

<a> * @param num 生産産品的數量 </a>

<a> public void produce(int num) </a>

<a> // 同步代碼段 </a>

<a> synchronized (list) </a>

<a> // 如果倉庫剩餘容量不足 </a>

<a> while (list.size() + num > max_size) </a>

<a> system.out.println("【要生産的産品數量】:" + num + " \t 【庫存量】:" </a>

<a> + list.size() + "\t 暫時不能執行生産任務!"); </a>

<a> // 由于條件不滿足，生産阻塞 </a>

<a> catch (interruptedexception e) </a>

<a> e.printstacktrace(); </a>

<a> // 生産條件滿足情況下，生産num個産品 </a>

<a> list.add(new object()); </a>

<a> system.out.println("【已經生産産品數】:" + num + "\t 【現倉儲量為】:" + list.size()); </a>

<a> list.notifyall(); </a>

<a> * 消費num個産品 </a>

<a> * @param num 消費産品數量 </a>

<a> public void consume(int num) </a>

<a> // 如果倉庫存儲量不足 </a>

<a> while (list.size() < num) </a>

<a> system.out.println("【要消費的産品數量】:" + num + " \t【庫存量】:" </a>

<a> + list.size() + " \t 暫時不能執行生産任務!"); </a>

<a> // 由于條件不滿足，消費阻塞 </a>

<a> // 消費條件滿足情況下，消費num個産品 </a>

<a> list.remove(); </a>

<a> system.out.println("【已經消費産品數】:" + num + " \t 【現倉儲量為】:" + list.size()); </a>

<a> // get/set方法 </a>

<a> public linkedlist<object> getlist() </a>

<a> return list; </a>

<a> public void setlist(linkedlist<object> list) </a>

<a> public int getmax_size() </a>

<a> return max_size; </a>

2、生産者

/**

* 生産者類producer繼承線程類thread

* @author zcr

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)

}

3、消費者

* 消費者類consumer繼承線程類thread

public class consumer extends thread

// 每次消費的産品數量

public consumer(storage storage)

consume(num);

public void consume(int num)

storage.consume(num);

4、測試類

* 測試類test

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();

5、結果：

【要消費的産品數量】:50 【庫存量】:0 暫時不能執行生産任務!

【要消費的産品數量】:20 【庫存量】:0 暫時不能執行生産任務!

【已經生産産品數】:10 【現倉儲量為】:10

【要消費的産品數量】:20 【庫存量】:10 暫時不能執行生産任務!

【要消費的産品數量】:50 【庫存量】:10 暫時不能執行生産任務!

【已經生産産品數】:10 【現倉儲量為】:20

【已經生産産品數】:10 【現倉儲量為】:30

【要消費的産品數量】:50 【庫存量】:30 暫時不能執行生産任務!

【已經消費産品數】:20 【現倉儲量為】:10

【已經生産産品數】:10 【現倉儲量為】:40

【已經消費産品數】:30 【現倉儲量為】:10

【已經生産産品數】:80 【現倉儲量為】:90

【已經消費産品數】:50 【現倉儲量為】:40

看完上述代碼，對wait() / notify()方法實作的同步有了了解。你可能會對storage類中為什麼要定義public void produce(int num);和public void consume(int num);方法感到不解，為什麼不直接在生産者類producer和消費者類consumer中實作這兩個方法，卻要調用storage類中的實作呢？淡定，後文會有解釋。我們先往下走。

二、await() / signal()方法

在jdk5.0之後，java提供了更加健壯的線程處理機制，包括同步、鎖定、線程池等，它們可以實作更細粒度的線程控制。await()和

signal()就是其中用來做同步的兩種方法，它們的功能基本上和wait() / nofity()相同，完全可以取代它們，但是它們和新引入的鎖定機制lock直接挂鈎，具有更大的靈活性。通過在lock對象上調用newcondition()方法，将條件變量和一個鎖對象進行綁定，進而控制并發程式通路競争資源的安全。

下面來看代碼：

隻需要更新倉庫類storage的代碼即可，生産者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實作緩沖區

* @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)

// 如果倉庫存儲量不足

while (list.size() < num)

system.out.println("【要消費的産品數量】:" + num + "/t【庫存量】:" + list.size()

// 由于條件不滿足，消費阻塞

empty.await();

// 消費條件滿足情況下，消費num個産品

list.remove();

system.out.println("【已經消費産品數】:" + num + "/t【現倉儲量為】:" + list.size());

// set/get方法

public int getmax_size()

return max_size;

public linkedlist<object> getlist()

return list;

public void setlist(linkedlist<object> list)

this.list = list;

結果：

<a>【要消費的産品數量】:50/t【庫存量】:0/t暫時不能執行生産任務! </a>

<a>【已經生産産品數】:10/t【現倉儲量為】:10 </a>

<a>【已經生産産品數】:10/t【現倉儲量為】:20 </a>

<a>【要消費的産品數量】:30/t【庫存量】:20/t暫時不能執行生産任務! </a>

<a>【已經消費産品數】:20/t【現倉儲量為】:0 </a>

<a>【要消費的産品數量】:50/t【庫存量】:20/t暫時不能執行生産任務! </a>

<a>【已經生産産品數】:10/t【現倉儲量為】:30 </a>

<a>【已經生産産品數】:10/t【現倉儲量為】:40 </a>

<a>【要生産的産品數量】:80/t【庫存量】:40/t暫時不能執行生産任務! </a>

<a>【已經消費産品數】:30/t【現倉儲量為】:10 </a>

<a>【要消費的産品數量】:50/t【庫存量】:10/t暫時不能執行生産任務! </a>

<a>【已經生産産品數】:80/t【現倉儲量為】:90 </a>

<a>【已經消費産品數】:50/t【現倉儲量為】:40</a>

這樣我們就知道為神馬我要在storage類中定義public void produce(int num);和public void consume(int num);方法，并在生産者類producer和消費者類consumer中調用storage類中的實作了吧。将可能發生的變化集中到一個類中，不影響原有的構架設計，同時無需修改其他業務層代碼。

總結

兩種方式原理一緻，都是對獨占空間加鎖，阻塞和喚醒線程，第一種方式比較傳統，第二種方式速度比較快。

緻謝：感謝您的耐心閱讀！

作者：whywin

來源：51cto

java 生産者消費者問題

繼續閱讀

關于Gradle配置的小結

Java小案例——随機數猜測随機數猜測

nginx location中斜線的位置的重要性

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的簡單使用

neo4j之cypher使用文檔

GitHub連夜封殺！這份阿裡 10W 字内部 Java 字面試手冊到底有多強？

spark/scala關于【資源檔案】加載方法概述外部檔案加載方案測試資源檔案打包入jar包中小結

mybatis_入門程式Mybatis入門

AOP程式設計_Android優雅權限架構(1)概念基礎，2021金三銀四前言正文大綱正文

Effective Java 8:通用程式設計

OOM三種類型

工廠模式-三種類型

【遞歸】高效率求2的n次幂

win10本地scala和spark安裝安裝scala安裝spark

scala (3) Function 和 Method