生產者消費者問題是研究多線程程序時繞不開的問題，它的描述是有一塊生產者和消費者共享的有界緩沖區(qū)，生產者往緩沖區(qū)放入產品，消費者從緩沖區(qū)取走產品，這個過程可以無休止的執(zhí)行，不能因緩沖區(qū)滿生產者放不進產品而終止，也不能因緩沖區(qū)空消費者無產品可取而終止。
???????
解決生產者消費者問題的方法有兩種，一種是采用某種機制保持生產者和消費者之間的同步，一種是在生產者和消費者之間建立一個管道。前一種有較高的效率并且可控制性較好，比較常用，后一種由于管道緩沖區(qū)不易控制及被傳輸數(shù)據(jù)對象不易封裝等原因，比較少用。
???????
同步問題的核心在于，CPU是按時間片輪詢的方式執(zhí)行程序，我們無法知道某一個線程是否被執(zhí)行、是否被搶占、是否結束等，因此生產者完全可能當緩沖區(qū)已滿的時候還在放入產品，消費者也完全可能當緩沖區(qū)為空時還在取出產品。
???????
現(xiàn)在同步問題的解決方法一般是采用信號或者加鎖機制，即生產者線程當緩沖區(qū)已滿時放棄自己的執(zhí)行權，進入等待狀態(tài)，并通知消費者線程執(zhí)行。消費者線程當緩沖區(qū)已空時放棄自己的執(zhí)行權，進入等待狀態(tài)，并通知生產者線程執(zhí)行。這樣一來就保持了線程的同步，并避免了線程間互相等待而進入死鎖狀態(tài)。
?????? JAVA
語言提供了獨立于平臺的線程機制，保持了”write once, run anywhere”的特色。同時也提供了對同步機制的良好支持。
???????
在JAVA中，一共有四種方法支持同步，其中三個是同步方法，一個是管道方法。
1.
??????
?
方法
wait()/notify
()
2.
???????
方法
await()/signal()
3.
???????
阻塞隊列方法
BlockingQueue
4.
???????
管道方法
PipedInputStream/PipedOutputStream
下面我們看各個方法的實現(xiàn)：
1.
???????
方法
wait()/notify()
wait()和notify()是根類Object的兩個方法，也就意味著所有的JAVA類都會具有這個兩個方法，為什么會被這樣設計呢？我們可以認為所有的對象默認都具有一個鎖，雖然我們看不到，也沒有辦法直接操作，但它是存在的。
wait()方法表示：當緩沖區(qū)已滿或空時，生產者或消費者線程停止自己的執(zhí)行，放棄鎖，使自己處于等待狀態(tài)，讓另一個線程開始執(zhí)行；
notify()方法表示：當生產者或消費者對緩沖區(qū)放入或取出一個產品時，向另一個線程發(fā)出可執(zhí)行通知，同時放棄鎖，使自己處于等待狀態(tài)。
下面是一個例子代碼：
import
?java.util.LinkedList;
public
?
class
?Sycn1
{
????
private
?LinkedList
<
Object
>
?myList?
=
new
?LinkedList
<
Object
>
();
????
private
?
int
?MAX?
=
?
10
;
????
????
public
?Sycn1()
{
????}
????
????
public
?
void
?start()
{
????????????
new
?Producer().start();
????????????
new
?Consumer().start();
????}
????
????
public
?
static
?
void
?main(String[]?args)?
throws
?Exception
{
????????Sycn1?s1?
=
?
new
?Sycn1();
????????s1.start();
????}
????
????
class
?Producer?
extends
?Thread
{????????
????????
public
?
void
?run()
{
????????????
while
(
true
)
{
????????????????
synchronized
(myList)
{
????????????????????
try
{
????????????????????????
while
(myList.size()?
==
?MAX)
{
????????????????????????????System.out.println(
"
warning:?it's?full!
"
);
????????????????????????????myList.wait();
????????????????????????}
????????????????????????Object?o?
=
?
new
?Object();
????????????????????????
if
(myList.add(o))
{
????????????????????????????System.out.println(
"
Producer:?
"
?
+
?o);
????????????????????????????myList.notify();
????????????????????????}
????????????????????}
catch
(InterruptedException?ie)
{
????????????????????????System.out.println(
"
producer?is?interrupted!
"
);
????????????????????}
????????????????}
????????????}
????????}
????}
????
????
class
?Consumer?
extends
?Thread
{
????????
public
?
void
?run()
{
????????????
while
(
true
)
{
????????????????
synchronized
(myList)
{
????????????????????
try
{
????????????????????????
while
(myList.size()?
==
?
0
)
{
????????????????????????????System.out.println(
"
warning:?it's?empty!
"
);
????????????????????????????myList.wait();
????????????????????????}
????????????????????????Object?o?
=
?myList.removeLast();
????????????????????????System.out.println(
"
Consumer:?
"
?
+
?o);
????????????????????????myList.notify();
????????????????????}
catch
(InterruptedException?ie)
{
????????????????????????System.out.println(
"
consumer?is?interrupted!
"
);
????????????????????}
????????????????}
????????????}
????????}
????}
????
}
2.
???????
方法
await()/signal()
在JDK5.0以后，JAVA提供了新的更加健壯的線程處理機制，包括了同步、鎖定、線程池等等，它們可以實現(xiàn)更小粒度上的控制。await()和signal()就是其中用來做同步的兩種方法，它們的功能基本上和wait()/notify()相同，完全可以取代它們，但是它們和新引入的鎖定機制Lock直接掛鉤，具有更大的靈活性。
下面是一個例子代碼：
import
?java.util.LinkedList;
import
?java.util.concurrent.locks.
*
;
public
?
class
?Sycn2
{
????
private
?LinkedList
<
Object
>
?myList?
=
?
new
?LinkedList
<
Object
>
();
????
private
?
int
?MAX?
=
?
10
;
????
private
?
final
?Lock?lock?
=
?
new
?ReentrantLock();
????
private
?
final
?Condition?full?
=
?lock.newCondition();
????
private
?
final
?Condition?empty?
=
?lock.newCondition();
????
????
public
?Sycn2()
{
????}
????
????
public
?
void
?start()
{
????????????
new
?Producer().start();
????????????
new
?Consumer().start();
????}
????
????
public
?
static
?
void
?main(String[]?args)?
throws
?Exception
{
????????Sycn2?s2?
=
?
new
?Sycn2();
????????s2.start();
????}
????
????
class
?Producer?
extends
?Thread
{????????
????????
public
?
void
?run()
{
????????????
while
(
true
)
{
????????????????lock.lock();
????????????????
try
{
????????????????????
while
(myList.size()?
==
?MAX)
{
????????????????????????System.out.println(
"
warning:?it's?full!
"
);
????????????????????????full.await();
????????????????????}
????????????????????Object?o?
=
?
new
?Object();
????????????????????
if
(myList.add(o))
{
????????????????????????System.out.println(
"
Producer:?
"
?
+
?o);
????????????????????????empty.signal();
????????????????????}
????????????????}
catch
(InterruptedException?ie)
{
????????????????????System.out.println(
"
producer?is?interrupted!
"
);
????????????????}
finally
{
????????????????????lock.unlock();
????????????????}
????????????}
????????}
????}
????
????
class
?Consumer?
extends
?Thread
{
????????
public
?
void
?run()
{
????????????
while
(
true
)
{
????????????????lock.lock();
????????????????
try
{
????????????????????
while
(myList.size()?
==
?
0
)
{
????????????????????????System.out.println(
"
warning:?it's?empty!
"
);
????????????????????????empty.await();
????????????????????}
????????????????????Object?o?
=
?myList.removeLast();
????????????????????System.out.println(
"
Consumer:?
"
?
+
?o);
????????????????????full.signal();
????????????????}
catch
(InterruptedException?ie)
{
????????????????????System.out.println(
"
consumer?is?interrupted!
"
);
????????????????}
finally
{
????????????????????lock.unlock();
????????????????}
????????????}
????????}
????}
????
}
3.
???????
阻塞隊列方法
BlockingQueue
BlockingQueue也是JDK5.0的一部分，它是一個已經在內部實現(xiàn)了同步的隊列，實現(xiàn)方式采用的是我們的第2種await()/signal()方法。它可以在生成對象時指定容量大小。
它用于阻塞操作的是put()和take()方法。
put()方法類似于我們上面的生產者線程，容量最大時，自動阻塞。
take()方法類似于我們上面的消費者線程，容量為0時，自動阻塞。
下面是一個例子代碼：
import
?java.util.concurrent.
*
;
public
?
class
?Sycn3
{
????
private
?LinkedBlockingQueue
<
Object
>
?queue?
=
?
new
?LinkedBlockingQueue
<
Object
>
(
10
);
????
private
?
int
?MAX?
=
?
10
;
????
????
public
?Sycn3()
{
????}
????
????
public
?
void
?start()
{
????????????
new
?Producer().start();
????????????
new
?Consumer().start();
????}
????
????
public
?
static
?
void
?main(String[]?args)?
throws
?Exception
{
????????Sycn3?s3?
=
?
new
?Sycn3();
????????s3.start();
????}
????
????
class
?Producer?
extends
?Thread
{????????
????????
public
?
void
?run()
{
????????????
while
(
true
)
{
????????????????
//
synchronized(this){
????????????????
try
{
????????????????????
if
(queue.size()?
==
?MAX)
????????????????????????System.out.println(
"
warning:?it's?full!
"
);
????????????????????Object?o?
=
?
new
?Object();
????????????????????queue.put(o);
????????????????????System.out.println(
"
Producer:?
"
?
+
?o);
????????????????????}
catch
(InterruptedException?e)
{
????????????????????????System.out.println(
"
producer?is?interrupted!
"
);
????????????????????}
????????????????
//
}
????????????}
????????}
????}
????
????
class
?Consumer?
extends
?Thread
{
????????
public
?
void
?run()
{
????????????
while
(
true
)
{
????????????????
//
synchronized(this){
????????????????
try
{
????????????????????
if
(queue.size()?
==
?
0
)
????????????????????????System.out.println(
"
warning:?it's?empty!
"
);
????????????????????Object?o?
=
?queue.take();
????????????????????System.out.println(
"
Consumer:?
"
?
+
?o);
????????????????????}
catch
(InterruptedException?e)
{
????????????????????????System.out.println(
"
producer?is?interrupted!
"
);
????????????????????}
????????????????
//
}
????????????}
????????}
????}
????
}
你發(fā)現(xiàn)這個例子中的問題了嗎？
如果沒有，我建議你運行一下這段代碼，仔細觀察它的輸出，是不是有下面這個樣子的？為什么會這樣呢？
…
warning: it's full!
Producer: java.lang.object@4526e2a
…
你可能會說這是因為put()和System.out.println()之間沒有同步造成的，我也這樣認為，我也這樣認為，但是你把run()中的synchronized前面的注釋去掉，重新編譯運行，有改觀嗎？沒有。為什么？
這是因為，當緩沖區(qū)已滿，生產者在put()操作時，put()內部調用了await()方法，放棄了線程的執(zhí)行，然后消費者線程執(zhí)行，調用take()方法，take()內部調用了signal()方法，通知生產者線程可以執(zhí)行，致使在消費者的println()還沒運行的情況下生產者的println()先被執(zhí)行，所以有了上面的輸出。run()中的synchronized其實并沒有起什么作用。
對于BlockingQueue大家可以放心使用，這可不是它的問題，只是在它和別的對象之間的同步有問題。
對于這種多重嵌套同步的問題，以后再談吧，歡迎大家討論?。?
4.
???????
管道方法
PipedInputStream/PipedOutputStream
這個類位于java.io包中，是解決同步問題的最簡單的辦法，一個線程將數(shù)據(jù)寫入管道，另一個線程從管道讀取數(shù)據(jù)，這樣便構成了一種生產者/消費者的緩沖區(qū)編程模式。
下面是一個例子代碼，在這個代碼我沒有使用Object對象，而是簡單的讀寫字節(jié)值，這是因為PipedInputStream/PipedOutputStream不允許傳輸對象，這是JAVA本身的一個bug,具體的大家可以看sun的解釋：
http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=4131126
import
?java.io.
*
;
public
?
class
?Sycn4
{
????
private
?PipedOutputStream?pos;
????
private
?PipedInputStream?pis;
????
//
private?ObjectOutputStream?oos;
????
//
private?ObjectInputStream?ois;
????
????
public
?Sycn4()
{
????????
try
{
????????????pos?
=
?
new
?PipedOutputStream();
????????????pis?
=
?
new
?PipedInputStream(pos);
????????????
//
oos?=?new?ObjectOutputStream(pos);
????????????
//
ois?=?new?ObjectInputStream(pis);
????????}
catch
(IOException?e)
{
????????????System.out.println(e);
????????}
????}
????
????
public
?
void
?start()
{
????????
new
?Producer().start();
????????
new
?Consumer().start();
????}
????
????
public
?
static
?
void
?main(String[]?args)?
throws
?Exception
{
????????Sycn4?s4?
=
?
new
?Sycn4();
????????s4.start();
????}
????
????
class
?Producer?
extends
?Thread
{
????????
public
?
void
?run()?
{
????????????
try
{
????????????????
while
(
true
)
{
????????????????????
int
?b?
=
?(
int
)?(Math.random()?
*
?
255
);
????????????????????System.out.println(
"
Producer:?a?byte,?the?value?is?
"
?
+
?b);
????????????????????pos.write(b);
????????????????????pos.flush();
????????????????????
//
Object?o?=?new?MyObject();
????????????????????
//
oos.writeObject(o);
????????????????????
//
oos.flush();
????????????????????
//
System.out.println("Producer:?"?+?o);
????????????????}
????????????}
catch
(Exception?e)
{
????????????????
//
System.out.println(e);
????????????????e.printStackTrace();
????????????}
finally
{
????????????????
try
{
????????????????????pos.close();
????????????????????pis.close();
????????????????????
//
oos.close();
????????????????????
//
ois.close();
????????????????}
catch
(IOException?e)
{
????????????????????System.out.println(e);
????????????????}
????????????}
????????}
????}
????
????
class
?Consumer?
extends
?Thread
{
????????
public
?
void
?run()
{
????????????
try
{
????????????????
while
(
true
)
{
????????????????????
int
?b?
=
?pis.read();
????????????????????System.out.println(
"
Consumer:?a?byte,?the?value?is?
"
?
+
?String.valueOf(b));
????????????????????
//
Object?o?=?ois.readObject();
????????????????????
//
if(o?!=?null)
????????????????????????
//
System.out.println("Consumer:?"?+?o);
????????????????}
????????????}
catch
(Exception?e)
{
????????????????
//
System.out.println(e);
????????????????e.printStackTrace();
????????????}
finally
{
????????????????
try
{
????????????????????pos.close();
????????????????????pis.close();
????????????????????
//
oos.close();
????????????????????
//
ois.close();
????????????????}
catch
(IOException?e)
{
????????????????????System.out.println(e);
????????????????}
????????????}
????????}
????}
????
????
//
class?MyObject?implements?Serializable?{
????
//
}
}
更多文章、技術交流、商務合作、聯(lián)系博主
微信掃碼或搜索：z360901061
微信掃一掃加我為好友
QQ號聯(lián)系： 360901061
您的支持是博主寫作最大的動力，如果您喜歡我的文章，感覺我的文章對您有幫助，請用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧，狠狠點擊下面給點支持吧，站長非常感激您！手機微信長按不能支付解決辦法：請將微信支付二維碼保存到相冊，切換到微信，然后點擊微信右上角掃一掃功能，選擇支付二維碼完成支付。
【本文對您有幫助就好】元
