[ Java設計模式 ] 多線程設計模式 : Producer-Consumer Pattern (我來做, 你來用)

前言 : 

這一章要學習的是 Producer-Consumer Pattern. Producer 是 “生產者” 的意思, 是指產生數據的線程. 而 Consumer 是 “消費者” 的意思, 意指使用數據的線程. 生產者必須將數據安全的交給消費者. 雖然聽起來簡單, 但當生產者與消費者在不同的線程上運行時, 兩者處理的速度差將是最大問題. 當消費者要取數據時, 生產者還沒建立數據, 或是生產者產出數據, 但是消費者還沒辦法接受數據等等. Producer-Consumer Pattern 是在生產者與消費者間加入一個 “橋樑參予者”, 這個橋樑用來緩衝兩者間的速度差.

Pattern 參與者 : 
* Data 參予者 : 

Data 參予者是由 Producer 參予者所建立, 並由 Consumer參予者所使用. 在以下範例, Data 參予者是 String 類(蛋糕).

* Producer 參予者 : 

Producer 參予者會建立 Data 參予者, 傳遞給 Channel 參予者. 在下面範例中由 MakerThread 擔任 Producer 參予者的角色.

* Consumer 參予者 : 

Consumer 參予者會利用 Channel 參予者獲取 Data 參予者. 在接下來範例由 EaterThread 擔任 Consumer 參予者角色.

* Channel 參予者 : 

Channel 參予者會從 Producer 參予者接收 Data 參予者, 並保管起來, 並回應Consumer 參予者要求, 將Data 參予者傳送出去. 為了確保Thread Safe Issue, Producer 參予者與 Consumer 參予者要對其進行訪問動作進行共享排斥

示意圖 : 
 

使用範例 : 

為了能了解 Producer-Consumer Pattern, 就先來看看範例程式. 這個範例程式中, 會有3 個廚師一直做蛋糕放在桌上, 而有三位客人則會不停著吃. 程序行為如下 :
* 廚師 (MakerThread) 會做蛋糕 (String) 並放在桌上 (Table).
* 桌上頂多可以放 3 個蛋糕.
* 桌上放滿 3個蛋糕, 而廚師還要放蛋糕, 需要等到桌上出現空位為止.
* 客人 (EaterThread) 會拿桌上的蛋糕吃.
* 客人會依次拿桌上蛋糕吃.
* 桌上沒有蛋糕時, 客人會等到桌上有蛋糕為止.

* Main 類 : 
Main 類會建立 Table, MakerThread 與 EaterThread 物件並啟動它們. 代碼如下 : 

view plaincopy to clipboardprint?

1. package dp.thread.ch5;  
2.   
3. public class Main {  
4.       
5.     public static void main(String args[]){  
6.         Table table = new Table(3); //建立可以放 3 個蛋糕的桌子  
7.         new MakerThread("MakerThread-1", table, 123231,5).start();  
8.         new MakerThread("MakerThread-2", table, 456654,5).start();  
9.         new MakerThread("MakerThread-3", table, 789987,5).start();  
10.         new EaterThread("EaterThread-1", table, 123321,5).start();  
11.         new EaterThread("EaterThread-2", table, 456654,5).start();  
12.         new EaterThread("EaterThread-3", table, 789987,5).start();        
13.     }  
14. }  

* MakerThread 類 : 
MakerThread 類是用來製作蛋糕並放在桌上的類, 也就是廚師. 代碼如下 : 

view plaincopy to clipboardprint?

1. package dp.thread.ch5;  
2.   
3. public class Main {  
4.       
5.     public static void main(String args[]){  
6.         Table table = new Table(3); //建立可以放 3 個蛋糕的桌子  
7.         new MakerThread("MakerThread-1", table, 123231,5).start();  
8.         new MakerThread("MakerThread-2", table, 456654,5).start();  
9.         new MakerThread("MakerThread-3", table, 789987,5).start();  
10.         new EaterThread("EaterThread-1", table, 123321,5).start();  
11.         new EaterThread("EaterThread-2", table, 456654,5).start();  
12.         new EaterThread("EaterThread-3", table, 789987,5).start();        
13.     }  
14. }  

* EaterThread 類 : 
EaterThread 類用來表示從桌上拿蛋糕來吃的客人. 桌上的蛋糕可以使用 Table 類的 take 方法獲取. 接者 MakerThread 會隨機暫停模擬 "吃蛋糕所花費的時間". 代碼如下 : 

view plaincopy to clipboardprint?

1. package dp.thread.ch5;  
2.   
3. import java.util.*;  
4.   
5. public class EaterThread extends Thread{  
6.     private final Random random;  
7.     private final Table table;  
8.     private int taskCount = 10;  
9.       
10.     public EaterThread(String name, Table t, long seed, int tc){  
11.         super(name);  
12.         this.table = t;  
13.         this.random = new Random(seed);  
14.         if(tc>0){  
15.             taskCount = tc;  
16.         }         
17.     }  
18.       
19.     public void run(){  
20.         try{  
21.             for(int i=0;i
22.                 String cake = table.take();  
23.                 //System.out.println("Cake:"+cake+" is taken by "+getName());  
24.                 Thread.sleep(random.nextInt(1500));  
25.             }  
26.         }catch(InterruptedException e){  
27.             e.printStackTrace();  
28.         }  
29.     }     
30. }  

* Table 類 : 
Table 類代表用來放置蛋糕的桌子. 蛋糕可以放的數目由該建構子傳入參數決定. 代碼如下 : 

view plaincopy to clipboardprint?

1. package dp.thread.ch5;  
2.   
3. public class Table {  
4.     private final String[] buffer;  
5.     private int tail;  //下一個put 的位址  
6.     private int head;  //下一個take 的地方  
7.     private int count; //buffer 內的蛋糕數  
8.       
9.     public Table(int count){  
10.         this.buffer = new String[count];  
11.         this.head = 0;  
12.         this.tail = 0;  
13.         this.count = 0;  
14.     }  
15.       
16.     public synchronized void put(String c) throws InterruptedException{  
17.         System.out.println(Thread.currentThread().getName()+ " puts "+c);  
18.         while(count >= buffer.length){  
19.             wait();  
20.         }  
21.         buffer[tail] = c;  
22.         tail = (tail + 1) % buffer.length;  
23.         count++;  
24.         notifyAll();  
25.     }  
26.       
27.     public synchronized String take() throws InterruptedException{  
28.         while(count <= 0){  
29.             wait();  
30.         }         
31.         String cake = buffer[head];  
32.         head = ( head+1 ) % buffer.length;  
33.         count--;  
34.         notifyAll();  
35.         System.out.println(Thread.currentThread().getName()+ " takes "+cake);  
36.         return cake;  
37.     }  
38. }  

執行結果 : 

MakerThread-2 puts [Cake No.0 made by MakerThread-2]
EaterThread-2 takes [Cake No.0 made by MakerThread-2]
MakerThread-2 puts [Cake No.1 made by MakerThread-2]
EaterThread-2 takes [Cake No.1 made by MakerThread-2]
MakerThread-2 puts [Cake No.2 made by MakerThread-2]
...(中間省略)...
EaterThread-1 takes [Cake No.12 made by MakerThread-1]
EaterThread-3 takes [Cake No.13 made by MakerThread-3]
EaterThread-3 takes [Cake No.14 made by MakerThread-1]

補充說明 : 
@. 在Table 類中的 put 方法中使用了 Guarded Suspension Pattern. 而警戒條件則是 while 條件的邏輯判斷, 如下所示 : 

view plaincopy to clipboardprint?

1. while(count >= buffer.length){ //若蛋糕數大於等於, Table 所能放的數目則暫停線程.  
2.             wait();  
3. }  

@. Table 類的 take 方法也使用了Guarded Suspension Pattern. 而警戒條件如下 : 

view plaincopy to clipboardprint?

1. while(count <= 0){ //若桌上已沒有蛋糕, 則暫停線程  
2.     wait();  
3. }  

@. 保護安全性的Channel : 

Producer-Consumer Pattern 中, 肩負保護安全性的使命是Channel 參與者. Channel 參與者進行線程間的共享互斥, 讓Producer 參與者能正確的將Data 參與者送到 Consumer 參與者手上. 範例程式中, Table 類的 put 方法 與 take 方法都使用了 Guarded Suspension Pattern. 但 MakerThread 與 EaterThread 類都不想依賴於Table 的詳細實現. 也就是說 MakerThread 不必理會其它線程, 只要調用 put 方法就好; 而 EaterThread 也是只要調用 take 方法就好. 使用 synchronized, wiat 與 notifyAll 這些考慮到多線程的操作代碼, 都已經封裝在 Channel 參與者 (Table)裡面.