一、概念
生產(chǎn)者與消費(fèi)者問題是一個金典的多線程協(xié)作的問題.生產(chǎn)者負(fù)責(zé)生產(chǎn)產(chǎn)品，并將產(chǎn)品存放到倉庫；消費(fèi)者從倉庫中獲取產(chǎn)品并消費(fèi)。當(dāng)倉庫滿時(shí)，生產(chǎn)者必須停止生產(chǎn)，直到倉庫有位置存放產(chǎn)品；當(dāng)倉庫空時(shí)，消費(fèi)者必須停止消費(fèi)，直到倉庫中有產(chǎn)品。
解決生產(chǎn)者/消費(fèi)者問題主要用到如下幾個技術(shù)：1.用線程模擬生產(chǎn)者，在run方法中不斷地往倉庫中存放產(chǎn)品。2.用線程模擬消費(fèi)者，在run方法中不斷地從倉庫中獲取產(chǎn)品。3
. 倉庫類保存產(chǎn)品，當(dāng)產(chǎn)品數(shù)量為0時(shí)，調(diào)用wait方法，使得當(dāng)前消費(fèi)者線程進(jìn)入等待狀態(tài)，當(dāng)有新產(chǎn)品存入時(shí)，調(diào)用notify方法，喚醒等待的消費(fèi)者線程。當(dāng)倉庫滿時(shí)，調(diào)用wait方法，使得當(dāng)前生產(chǎn)者線程進(jìn)入等待狀態(tài)，當(dāng)有消費(fèi)者獲取產(chǎn)品時(shí)，調(diào)用notify方法，喚醒等待的生產(chǎn)者線程。
二、實(shí)例
|
1
2
3
4
5
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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
|
package book.thread.product;public class Consumer extends Thread{ private Warehouse warehouse;//消費(fèi)者獲取產(chǎn)品的倉庫 private boolean running = false;//是否需要結(jié)束線程的標(biāo)志位 public Consumer(Warehouse warehouse,String name){ super(name); this.warehouse = warehouse; } public void start(){ this.running = true; super.start(); } public void run(){ Product product; try { while(running){ //從倉庫中獲取產(chǎn)品 product = warehouse.getProduct(); sleep(500); } } catch (InterruptedException e) { e.printStackTrace(); } } //停止消費(fèi)者線程 public void stopConsumer(){ synchronized(warehouse){ this.running = false; warehouse.notifyAll();//通知等待倉庫的線程 } } //消費(fèi)者線程是否在運(yùn)行 public boolean isRunning(){ return running; }} package book.thread.product;public class Producer extends Thread{ private Warehouse warehouse;//生產(chǎn)者存儲產(chǎn)品的倉庫 private static int produceName = 0;//產(chǎn)品的名字 private boolean running = false;//是否需要結(jié)束線程的標(biāo)志位 public Producer(Warehouse warehouse,String name){ super(name); this.warehouse = warehouse; } public void start(){ this.running = true; super.start(); } public void run(){ Product product; //生產(chǎn)并存儲產(chǎn)品 try { while(running){ product = new Product((++produceName)+""); this.warehouse.storageProduct(product); sleep(300); } } catch (InterruptedException e) { e.printStackTrace(); } } //停止生產(chǎn)者線程 public void stopProducer(){ synchronized(warehouse){ this.running = false; //通知等待倉庫的線程 warehouse.notifyAll(); } } //生產(chǎn)者線程是否在運(yùn)行 public boolean isRunning(){ return running; }} package book.thread.product;public class Product { private String name;//產(chǎn)品名 public Product(String name){ this.name = name; } public String toString(){ return "Product-"+name; }} package book.thread.product;//產(chǎn)品的倉庫類，內(nèi)部采用數(shù)組來表示循環(huán)隊(duì)列，以存放產(chǎn)品public class Warehouse { private static int CAPACITY = 11;//倉庫的容量 private Product[] products;//倉庫里的產(chǎn)品 //[front,rear]區(qū)間的產(chǎn)品未被消費(fèi) private int front = 0;//當(dāng)前倉庫中第一個未被消費(fèi)的產(chǎn)品的下標(biāo) private int rear = 0;//倉庫中最后一個未被消費(fèi)的產(chǎn)品下標(biāo)加1 public Warehouse(){ this.products = new Product[CAPACITY]; } public Warehouse(int capacity){ this(); if(capacity > 0){ CAPACITY = capacity +1; this.products = new Product[CAPACITY]; } } //從倉庫獲取一個產(chǎn)品 public Product getProduct() throws InterruptedException{ synchronized(this){ boolean consumerRunning = true;//標(biāo)志消費(fèi)者線程是否還在運(yùn)行 Thread currentThread = Thread.currentThread();//獲取當(dāng)前線程 if(currentThread instanceof Consumer){ consumerRunning = ((Consumer)currentThread).isRunning(); }else{ return null;//非消費(fèi)者不能獲取產(chǎn)品 } //若消費(fèi)者線程在運(yùn)行中，但倉庫中沒有產(chǎn)品了，則消費(fèi)者線程繼續(xù)等待 while((front==rear) && consumerRunning){ wait(); consumerRunning = ((Consumer)currentThread).isRunning(); } //如果消費(fèi)者線程已經(jīng)停止運(yùn)行，則退出該方法，取消獲取產(chǎn)品 if(!consumerRunning){ return null; } //獲取當(dāng)前未被消費(fèi)的第一個產(chǎn)品 Product product = products[front]; System.out.println("Consumer[" + currentThread.getName()+"] getProduct:"+product); //將當(dāng)前未被消費(fèi)產(chǎn)品的下標(biāo)后移一位，如果到了數(shù)組末尾，則移動到首部 front = (front+1+CAPACITY)%CAPACITY; System.out.println("倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量："+(rear+CAPACITY-front)%CAPACITY); //通知其他等待線程 notify(); return product; } } //向倉庫存儲一個產(chǎn)品 public void storageProduct(Product product) throws InterruptedException{ synchronized(this){ boolean producerRunning = true;//標(biāo)志生產(chǎn)者線程是否在運(yùn)行 Thread currentThread = Thread.currentThread(); if(currentThread instanceof Producer){ producerRunning = ((Producer)currentThread).isRunning(); }else{ return; } //如果最后一個未被消費(fèi)的產(chǎn)品與第一個未被消費(fèi)的產(chǎn)品的下標(biāo)緊挨著，則說明沒有存儲空間了。 //如果沒有存儲空間了，而生產(chǎn)者線程還在運(yùn)行，則生產(chǎn)者線程等待倉庫釋放產(chǎn)品 while(((rear+1)%CAPACITY == front) && producerRunning){ wait(); producerRunning = ((Producer)currentThread).isRunning(); } //如果生產(chǎn)線程已經(jīng)停止運(yùn)行了，則停止產(chǎn)品的存儲 if(!producerRunning){ return; } //保存產(chǎn)品到倉庫 products[rear] = product; System.out.println("Producer[" + Thread.currentThread().getName()+"] storageProduct:" + product); //將rear下標(biāo)循環(huán)后移一位 rear = (rear + 1)%CAPACITY; System.out.println("倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量："+(rear + CAPACITY -front)%CAPACITY); notify(); } }} package book.thread.product;public class TestProduct { public static void main(String[] args) { Warehouse warehouse = new Warehouse(10);//建立一個倉庫，容量為10 //建立生產(chǎn)者線程和消費(fèi)者 Producer producers1 = new Producer(warehouse,"producer-1"); Producer producers2 = new Producer(warehouse,"producer-2"); Producer producers3 = new Producer(warehouse,"producer-3"); Consumer consumer1 = new Consumer(warehouse,"consumer-1"); Consumer consumer2 = new Consumer(warehouse,"consumer-2"); Consumer consumer3 = new Consumer(warehouse,"consumer-3"); Consumer consumer4 = new Consumer(warehouse,"consumer-4"); //啟動生產(chǎn)者線程和消費(fèi)者線程 producers1.start(); producers2.start(); consumer1.start(); producers3.start(); consumer2.start(); consumer3.start(); consumer4.start(); //讓生產(chǎn)者/消費(fèi)者程序運(yùn)行1600ms try { Thread.sleep(1600); } catch (InterruptedException e) { e.printStackTrace(); } //停止消費(fèi)者線程 producers1.stopProducer(); consumer1.stopConsumer(); producers2.stopProducer(); consumer2.stopConsumer(); producers3.stopProducer(); consumer3.stopConsumer(); consumer4.stopConsumer(); }} |
輸出結(jié)果：
|
1
2
3
4
5
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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
|
Producer[producer-1] storageProduct:Product-1倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：1Consumer[consumer-2] getProduct:Product-1倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：0Producer[producer-3] storageProduct:Product-3倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：1Producer[producer-2] storageProduct:Product-2倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：2Consumer[consumer-3] getProduct:Product-3倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：1Consumer[consumer-1] getProduct:Product-2倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：0Producer[producer-1] storageProduct:Product-4倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：1Consumer[consumer-4] getProduct:Product-4倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：0Producer[producer-3] storageProduct:Product-6倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：1Producer[producer-2] storageProduct:Product-5倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：2Consumer[consumer-1] getProduct:Product-6倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：1Consumer[consumer-2] getProduct:Product-5倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：0Producer[producer-1] storageProduct:Product-7倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：1Consumer[consumer-3] getProduct:Product-7倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：0Producer[producer-3] storageProduct:Product-8倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：1Producer[producer-2] storageProduct:Product-9倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：2Consumer[consumer-4] getProduct:Product-8倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：1Producer[producer-1] storageProduct:Product-10倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：2Producer[producer-3] storageProduct:Product-11倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：3Producer[producer-2] storageProduct:Product-12倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：4Consumer[consumer-1] getProduct:Product-9倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：3Consumer[consumer-2] getProduct:Product-10倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：2Consumer[consumer-3] getProduct:Product-11倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：1Producer[producer-3] storageProduct:Product-13倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：2Producer[producer-1] storageProduct:Product-14倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：3Producer[producer-2] storageProduct:Product-15倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：4Consumer[consumer-4] getProduct:Product-12倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：3Consumer[consumer-1] getProduct:Product-13倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：2Consumer[consumer-2] getProduct:Product-14倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：1Producer[producer-1] storageProduct:Product-16倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：2Producer[producer-3] storageProduct:Product-17倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：3Producer[producer-2] storageProduct:Product-18倉庫中還沒有被消費(fèi)的產(chǎn)品數(shù)量：4 |
分析：在main方法中建立了一個產(chǎn)品倉庫，并未該倉庫關(guān)聯(lián)了3個生產(chǎn)者線程和4個消費(fèi)者線程，啟動這些線程，使生產(chǎn) 者/消費(fèi)者模型運(yùn)作起來，當(dāng)程序運(yùn)行1600ms時(shí)，所有的生產(chǎn)者停止生產(chǎn)產(chǎn)品，消費(fèi)者停止消費(fèi)產(chǎn)品。
生產(chǎn)者線程Product在run方法中沒300ms便生產(chǎn)一個產(chǎn)品，并存入倉庫；消費(fèi)者線程Consumer在run方法中沒500ms便從倉庫中取一個產(chǎn)品。
倉庫類Warehouse負(fù)責(zé)存放產(chǎn)品和發(fā)放產(chǎn)品。storageProduct方法負(fù)責(zé)存儲產(chǎn)品，當(dāng)倉庫滿時(shí)，當(dāng)前線程進(jìn)入等待狀態(tài)，即如果生產(chǎn)者線程A在調(diào)用storageProduct方法以存儲產(chǎn)品時(shí)，發(fā)現(xiàn)倉庫已滿，無法存儲時(shí)，便會進(jìn)入等待狀態(tài)。當(dāng)存儲產(chǎn)品成功時(shí)，調(diào)用notify方法，喚醒等待的消費(fèi)者線程。
getProduct方法負(fù)責(zé)提前產(chǎn)品，當(dāng)倉庫空時(shí)，當(dāng)前線程進(jìn)入等待狀態(tài)，即如果消費(fèi)者線程B在調(diào)用getProduct方法以獲取產(chǎn)品時(shí)，發(fā)現(xiàn)倉庫空了，便會進(jìn)入等待狀態(tài)。當(dāng)提取產(chǎn)品成功時(shí)，調(diào)用notify方法，喚醒等待的生產(chǎn)者線程。
以上這篇淺談java線程中生產(chǎn)者與消費(fèi)者的問題就是小編分享給大家的全部內(nèi)容了，希望能給大家一個參考，也希望大家多多支持服務(wù)器之家。
