【java】生产者/消费者模式--非阻塞队列

生产者消费者问题描述是有一块缓冲区做为仓库，生产者能够将产品放入仓库，消费者则能够从其取走产品。

解决问题分类

1. 同步：效率高易实现，代码控制性较好，经常使用；

2. 管道：不易控制，被传输数据对象不易于封装，实用性不强。（PipedInputStream/PipedOutputStream）

同步

（1）核心：保证同一资源被多个线程并发访问时的完整性。

（2）方法：

1. wait()/notify()

2. await()/signal()

3. BlockingQueue阻塞队列方法

wait()/notify()

是基类Object的两个方法，意味着全部Java类均可拥有，故可为任何对象实现同步机制。

wait()：当缓冲区已满/空时，对应线程中止本身的执行，放弃锁，等待；

notify()：线程执行后，向其余等待的线程发出可执行的通知，同时放弃锁，等待。

【代码】仓库使用时生产者与消费者同步synchronized（仓库），同步代码内根据条件（同步对象.wait&notify），不直接在生产者类Producer和消费者类Consumer中实现这两个方法，用Storage类中的实现：

1. import java.util.LinkedList;  
2.   
3. /** 
4.  * 仓库类Storage实现缓冲区 
5.  *  
6.  * Email:530025983@qq.com 
7.  *  
8.  * @author MONKEY.D.MENG 2011-03-15 
9.  *  
10.  */  
11. public class Storage  
12. {  
13.     // 仓库最大存储量  
14.     private final int MAX_SIZE = 100;  
15.   
16.     // 仓库存储的载体  
17.     private LinkedList<Object> list = new LinkedList<Object>();  
18.   
19.     // 生产num个产品  
20.     public void produce(int num)  
21.     {  
22.         // 同步代码段  
23.         synchronized (list)  
24.         {  
25.             // 若是仓库剩余容量不足  
26.             while (list.size() + num > MAX_SIZE)  
27.             {  
28.                 System.out.println("【要生产的产品数量】:" + num + "/t【库存量】:"  
29.                         + list.size() + "/t暂时不能执行生产任务!");  
30.                 try  
31.                 {  
32.                     // 因为条件不知足，生产阻塞  
33.                     list.wait();  
34.                 }  
35.                 catch (InterruptedException e)  
36.                 {  
37.                     e.printStackTrace();  
38.                 }  
39.             }  
40.   
41.             // 生产条件知足状况下，生产num个产品  
42.             for (int i = 1; i <= num; ++i)  
43.             {  
44.                 list.add(new Object());  
45.             }  
46.   
47.             System.out.println("【已经生产产品数】:" + num + "/t【现仓储量为】:" + list.size());  
48.   
49.             list.notifyAll();  
50.         }  
51.     }  
52.   
53.     // 消费num个产品  
54.     public void consume(int num)  
55.     {  
56.         // 同步代码段  
57.         synchronized (list)  
58.         {  
59.             // 若是仓库存储量不足  
60.             while (list.size() < num)  
61.             {  
62.                 System.out.println("【要消费的产品数量】:" + num + "/t【库存量】:"  
63.                         + list.size() + "/t暂时不能执行生产任务!");  
64.                 try  
65.                 {  
66.                     // 因为条件不知足，消费阻塞  
67.                     list.wait();  
68.                 }  
69.                 catch (InterruptedException e)  
70.                 {  
71.                     e.printStackTrace();  
72.                 }  
73.             }  
74.   
75.             // 消费条件知足状况下，消费num个产品  
76.             for (int i = 1; i <= num; ++i)  
77.             {  
78.                 list.remove();  
79.             }  
80.   
81.             System.out.println("【已经消费产品数】:" + num + "/t【现仓储量为】:" + list.size());  
82.   
83.             list.notifyAll();  
84.         }  
85.     }  
86.   
87.     // get/set方法  
88.     public LinkedList<Object> getList()  
89.     {  
90.         return list;  
91.     }  
92.   
93.     public void setList(LinkedList<Object> list)  
94.     {  
95.         this.list = list;  
96.     }  
97.   
98.     public int getMAX_SIZE()  
99.     {  
100.         return MAX_SIZE;  
101.     }  
102. }  
103. /** 
104.  * 生产者类Producer继承线程类Thread 
105.  *  
106.  * Email:530025983@qq.com 
107.  *  
108.  * @author MONKEY.D.MENG 2011-03-15 
109.  *  
110.  */  
111. public class Producer extends Thread  
112. {  
113.     // 每次生产的产品数量  
114.     private int num;  
115.   
116.     // 所在放置的仓库  
117.     private Storage storage;  
118.   
119.     // 构造函数，设置仓库  
120.     public Producer(Storage storage)  
121.     {  
122.         this.storage = storage;  
123.     }  
124.   
125.     // 线程run函数  
126.     public void run()  
127.     {  
128.         produce(num);  
129.     }  
130.   
131.     // 调用仓库Storage的生产函数  
132.     public void produce(int num)  
133.     {  
134.         storage.produce(num);  
135.     }  
136.   
137.     // get/set方法  
138.     public int getNum()  
139.     {  
140.         return num;  
141.     }  
142.   
143.     public void setNum(int num)  
144.     {  
145.         this.num = num;  
146.     }  
147.   
148.     public Storage getStorage()  
149.     {  
150.         return storage;  
151.     }  
152.   
153.     public void setStorage(Storage storage)  
154.     {  
155.         this.storage = storage;  
156.     }  
157. }  
158. /** 
159.  * 消费者类Consumer继承线程类Thread 
160.  *  
161.  * Email:530025983@qq.com 
162.  *  
163.  * @author MONKEY.D.MENG 2011-03-15 
164.  *  
165.  */  
166. public class Consumer extends Thread  
167. {  
168.     // 每次消费的产品数量  
169.     private int num;  
170.   
171.     // 所在放置的仓库  
172.     private Storage storage;  
173.   
174.     // 构造函数，设置仓库  
175.     public Consumer(Storage storage)  
176.     {  
177.         this.storage = storage;  
178.     }  
179.   
180.     // 线程run函数  
181.     public void run()  
182.     {  
183.         consume(num);  
184.     }  
185.   
186.     // 调用仓库Storage的生产函数  
187.     public void consume(int num)  
188.     {  
189.         storage.consume(num);  
190.     }  
191.   
192.     // get/set方法  
193.     public int getNum()  
194.     {  
195.         return num;  
196.     }  
197.   
198.     public void setNum(int num)  
199.     {  
200.         this.num = num;  
201.     }  
202.   
203.     public Storage getStorage()  
204.     {  
205.         return storage;  
206.     }  
207.   
208.     public void setStorage(Storage storage)  
209.     {  
210.         this.storage = storage;  
211.     }  
212. }  
213. /** 
214.  * <a href="http://lib.csdn.net/base/softwaretest" class='replace_word' title="软件测试知识库" target='_blank' style='color:#df3434; font-weight:bold;'>测试</a>类Test 
215.  *  
216.  * Email:530025983@qq.com 
217.  *  
218.  * @author MONKEY.D.MENG 2011-03-15 
219.  *  
220.  */  
221. public class Test  
222. {  
223.     public static void main(String[] args)  
224.     {  
225.         // 仓库对象  
226.         Storage storage = new Storage();  
227.   
228.         // 生产者对象  
229.         Producer p1 = new Producer(storage);  
230.         Producer p2 = new Producer(storage);  
231.         Producer p3 = new Producer(storage);  
232.         Producer p4 = new Producer(storage);  
233.         Producer p5 = new Producer(storage);  
234.         Producer p6 = new Producer(storage);  
235.         Producer p7 = new Producer(storage);  
236.   
237.         // 消费者对象  
238.         Consumer c1 = new Consumer(storage);  
239.         Consumer c2 = new Consumer(storage);  
240.         Consumer c3 = new Consumer(storage);  
241.   
242.         // 设置生产者产品生产数量  
243.         p1.setNum(10);  
244.         p2.setNum(10);  
245.         p3.setNum(10);  
246.         p4.setNum(10);  
247.         p5.setNum(10);  
248.         p6.setNum(10);  
249.         p7.setNum(80);  
250.   
251.         // 设置消费者产品消费数量  
252.         c1.setNum(50);  
253.         c2.setNum(20);  
254.         c3.setNum(30);  
255.   
256.         // 线程开始执行  
257.         c1.start();  
258.         c2.start();  
259.         c3.start();  
260.         p1.start();  
261.         p2.start();  
262.         p3.start();  
263.         p4.start();  
264.         p5.start();  
265.         p6.start();  
266.         p7.start();  
267.     }  
268. }  
269. 【要消费的产品数量】:50   【库存量】:0 暂时不能执行生产任务!  
270. 【要消费的产品数量】:30   【库存量】:0 暂时不能执行生产任务!  
271. 【要消费的产品数量】:20   【库存量】:0 暂时不能执行生产任务!  
272. 【已经生产产品数】:10    【现仓储量为】:10  
273. 【要消费的产品数量】:20   【库存量】:10    暂时不能执行生产任务!  
274. 【要消费的产品数量】:30   【库存量】:10    暂时不能执行生产任务!  
275. 【要消费的产品数量】:50   【库存量】:10    暂时不能执行生产任务!  
276. 【已经生产产品数】:10    【现仓储量为】:20  
277. 【要消费的产品数量】:50   【库存量】:20    暂时不能执行生产任务!  
278. 【要消费的产品数量】:30   【库存量】:20    暂时不能执行生产任务!  
279. 【已经消费产品数】:20    【现仓储量为】:0  
280. 【已经生产产品数】:10    【现仓储量为】:10  
281. 【已经生产产品数】:10    【现仓储量为】:20  
282. 【已经生产产品数】:80    【现仓储量为】:100  
283. 【要生产的产品数量】:10   【库存量】:100   暂时不能执行生产任务!  
284. 【已经消费产品数】:30    【现仓储量为】:70  
285. 【已经消费产品数】:50    【现仓储量为】:20  
286. 【已经生产产品数】:10    【现仓储量为】:30  
287. 【已经生产产品数】:10    【现仓储量为】:40  

await() / signal()

可彻底取代wait() / nofity()，但同时锁定机制Lock直接

挂钩，具备更大的灵活性。在Lock对象调用newCondition()方法，将条件变量和一个锁对象进行绑定，进而控制并发程序访问竞争资源的安全。

1. import java.util.LinkedList;  
2. import java.util.concurrent.locks.Condition;  
3. import java.util.concurrent.locks.Lock;  
4. import java.util.concurrent.locks.ReentrantLock;  
5.   
6. /** 
7.  * 仓库类Storage实现缓冲区 
8.  *  
9.  * Email:530025983@qq.com 
10.  *  
11.  * @author MONKEY.D.MENG 2011-03-15 
12.  *  
13.  */  
14. public class Storage  
15. {  
16.     // 仓库最大存储量  
17.     private final int MAX_SIZE = 100;  
18.   
19.     // 仓库存储的载体  
20.     private LinkedList<Object> list = new LinkedList<Object>();  
21.   
22.     // 锁  
23.     private final Lock lock = new ReentrantLock();  
24.   
25.     // 仓库满的条件变量  
26.     private final Condition full = lock.newCondition();  
27.   
28.     // 仓库空的条件变量  
29.     private final Condition empty = lock.newCondition();  
30.   
31.     // 生产num个产品  
32.     public void produce(int num)  
33.     {  
34.         // 得到锁  
35.         lock.lock();  
36.   
37.         // 若是仓库剩余容量不足  
38.         while (list.size() + num > MAX_SIZE)  
39.         {  
40.             System.out.println("【要生产的产品数量】:" + num + "/t【库存量】:" + list.size()  
41.                     + "/t暂时不能执行生产任务!");  
42.             try  
43.             {  
44.                 // 因为条件不知足，生产阻塞  
45.                 full.await();  
46.             }  
47.             catch (InterruptedException e)  
48.             {  
49.                 e.printStackTrace();  
50.             }  
51.         }  
52.   
53.         // 生产条件知足状况下，生产num个产品  
54.         for (int i = 1; i <= num; ++i)  
55.         {  
56.             list.add(new Object());  
57.         }  
58.   
59.         System.out.println("【已经生产产品数】:" + num + "/t【现仓储量为】:" + list.size());  
60.   
61.         // 唤醒其余全部线程  
62.         full.signalAll();  
63.         empty.signalAll();  
64.   
65.         // 释放锁  
66.         lock.unlock();  
67.     }  
68.   
69.     // 消费num个产品  
70.     public void consume(int num)  
71.     {  
72.         // 得到锁  
73.         lock.lock();  
74.   
75.         // 若是仓库存储量不足  
76.         while (list.size() < num)  
77.         {  
78.             System.out.println("【要消费的产品数量】:" + num + "/t【库存量】:" + list.size()  
79.                     + "/t暂时不能执行生产任务!");  
80.             try  
81.             {  
82.                 // 因为条件不知足，消费阻塞  
83.                 empty.await();  
84.             }  
85.             catch (InterruptedException e)  
86.             {  
87.                 e.printStackTrace();  
88.             }  
89.         }  
90.   
91.         // 消费条件知足状况下，消费num个产品  
92.         for (int i = 1; i <= num; ++i)  
93.         {  
94.             list.remove();  
95.         }  
96.   
97.         System.out.println("【已经消费产品数】:" + num + "/t【现仓储量为】:" + list.size());  
98.   
99.         // 唤醒其余全部线程  
100.         full.signalAll();  
101.         empty.signalAll();  
102.   
103.         // 释放锁  
104.         lock.unlock();  
105.     }  
106.   
107.     // set/get方法  
108.     public int getMAX_SIZE()  
109.     {  
110.         return MAX_SIZE;  
111.     }  
112.   
113.     public LinkedList<Object> getList()  
114.     {  
115.         return list;  
116.     }  
117.   
118.     public void setList(LinkedList<Object> list)  
119.     {  
120.         this.list = list;  
121.     }  
122. }  
123. 【要消费的产品数量】:50   【库存量】:0 暂时不能执行生产任务!  
124. 【要消费的产品数量】:30   【库存量】:0 暂时不能执行生产任务!  
125. 【已经生产产品数】:10    【现仓储量为】:10  
126. 【已经生产产品数】:10    【现仓储量为】:20  
127. 【要消费的产品数量】:50   【库存量】:20    暂时不能执行生产任务!  
128. 【要消费的产品数量】:30   【库存量】:20    暂时不能执行生产任务!  
129. 【已经生产产品数】:10    【现仓储量为】:30  
130. 【要消费的产品数量】:50   【库存量】:30    暂时不能执行生产任务!  
131. 【已经消费产品数】:20    【现仓储量为】:10  
132. 【已经生产产品数】:10    【现仓储量为】:20  
133. 【要消费的产品数量】:30   【库存量】:20    暂时不能执行生产任务!  
134. 【已经生产产品数】:80    【现仓储量为】:100  
135. 【要生产的产品数量】:10   【库存量】:100   暂时不能执行生产任务!  
136. 【已经消费产品数】:50    【现仓储量为】:50  
137. 【已经生产产品数】:10    【现仓储量为】:60  
138. 【已经消费产品数】:30    【现仓储量为】:30  
139. 【已经生产产品数】:10    【现仓储量为】:40  

只须要更新仓库类Storage的代码便可，生产者Producer、消费者Consumer、测试类Test的代码均不须要进行任何更改。这样咱们就知道为神马我要在Storage类中定义public void produce(int num);和public void consume(int num);方法，并在生产者类Producer和消费者类Consumer中调用Storage类中的实现了吧。将可能发生的变化集中到一个类中，不影响原有的构架设计，同时无需修改其余业务层代码。

BlockingQueue阻塞队列方法

BlockingQueue是JDK5.0的新增内容，它是一个已经在内部实现了同步的队列，实现方式采用的是咱们第2种await() / signal()方法。它能够在生成对象时指定容量大小。它用于阻塞操做的是put()和take()方法。

put()方法：相似于咱们上面的生产者线程，容量达到最大时，自动阻塞。

take()方法：相似于咱们上面的消费者线程，容量为0时，自动阻塞。

1. import java.util.concurrent.LinkedBlockingQueue;  
2.   
3. /** 
4.  * 仓库类Storage实现缓冲区 
5.  *  
6.  * Email:530025983@qq.com 
7.  *  
8.  * @author MONKEY.D.MENG 2011-03-15 
9.  *  
10.  */  
11. public class Storage  
12. {  
13.     // 仓库最大存储量  
14.     private final int MAX_SIZE = 100;  
15.   
16.     // 仓库存储的载体  
17.     private LinkedBlockingQueue<Object> list = new LinkedBlockingQueue<Object>(  
18.             100);  
19.   
20.     // 生产num个产品  
21.     public void produce(int num)  
22.     {  
23.         // 若是仓库剩余容量为0  
24.         if (list.size() == MAX_SIZE)  
25.         {  
26.             System.out.println("【库存量】:" + MAX_SIZE + "/t暂时不能执行生产任务!");  
27.         }  
28.   
29.         // 生产条件知足状况下，生产num个产品  
30.         for (int i = 1; i <= num; ++i)  
31.         {  
32.             try  
33.             {  
34.                 // 放入产品，自动阻塞  
35.                 list.put(new Object());  
36.             }  
37.             catch (InterruptedException e)  
38.             {  
39.                 e.printStackTrace();  
40.             }  
41.   
42.             System.out.println("【现仓储量为】:" + list.size());  
43.         }  
44.     }  
45.   
46.     // 消费num个产品  
47.     public void consume(int num)  
48.     {  
49.         // 若是仓库存储量不足  
50.         if (list.size() == 0)  
51.         {  
52.             System.out.println("【库存量】:0/t暂时不能执行生产任务!");  
53.         }  
54.   
55.         // 消费条件知足状况下，消费num个产品  
56.         for (int i = 1; i <= num; ++i)  
57.         {  
58.             try  
59.             {  
60.                 // 消费产品，自动阻塞  
61.                 list.take();  
62.             }  
63.             catch (InterruptedException e)  
64.             {  
65.                 e.printStackTrace();  
66.             }  
67.         }  
68.   
69.         System.out.println("【现仓储量为】:" + list.size());  
70.     }  
71.   
72.     // set/get方法  
73.     public LinkedBlockingQueue<Object> getList()  
74.     {  
75.         return list;  
76.     }  
77.   
78.     public void setList(LinkedBlockingQueue<Object> list)  
79.     {  
80.         this.list = list;  
81.     }  
82.   
83.     public int getMAX_SIZE()  
84.     {  
85.         return MAX_SIZE;  
86.     }  
87. }  
88. 【库存量】:0 暂时不能执行生产任务!  
89. 【库存量】:0 暂时不能执行生产任务!  
90. 【现仓储量为】:1  
91. 【现仓储量为】:1  
92. 【现仓储量为】:3  
93. 【现仓储量为】:4  
94. 【现仓储量为】:5  
95. 【现仓储量为】:6  
96. 【现仓储量为】:7  
97. 【现仓储量为】:8  
98. 【现仓储量为】:9  
99. 【现仓储量为】:10  
100. 【现仓储量为】:11  
101. 【现仓储量为】:1  
102. 【现仓储量为】:2  
103. 【现仓储量为】:13  
104. 【现仓储量为】:14  
105. 【现仓储量为】:17  
106. 【现仓储量为】:19  
107. 【现仓储量为】:20  
108. 【现仓储量为】:21  
109. 【现仓储量为】:22  
110. 【现仓储量为】:23  
111. 【现仓储量为】:24  
112. 【现仓储量为】:25  
113. 【现仓储量为】:26  
114. 【现仓储量为】:12  
115. 【现仓储量为】:1  
116. 【现仓储量为】:1  
117. 【现仓储量为】:2  
118. 【现仓储量为】:3  
119. 【现仓储量为】:4  
120. 【现仓储量为】:5  
121. 【现仓储量为】:6  
122. 【现仓储量为】:7  
123. 【现仓储量为】:27  
124. 【现仓储量为】:8  
125. 【现仓储量为】:6  
126. 【现仓储量为】:18  
127. 【现仓储量为】:2  
128. 【现仓储量为】:3  
129. 【现仓储量为】:4  
130. 【现仓储量为】:5  
131. 【现仓储量为】:6  
132. 【现仓储量为】:7  
133. 【现仓储量为】:8  
134. 【现仓储量为】:9  
135. 【现仓储量为】:10  
136. 【现仓储量为】:16  
137. 【现仓储量为】:11  
138. 【现仓储量为】:12  
139. 【现仓储量为】:13  
140. 【现仓储量为】:14  
141. 【现仓储量为】:15  
142. 【现仓储量为】:1  
143. 【现仓储量为】:2  
144. 【现仓储量为】:3  
145. 【现仓储量为】:3  
146. 【现仓储量为】:15  
147. 【现仓储量为】:1  
148. 【现仓储量为】:0  
149. 【现仓储量为】:1  
150. 【现仓储量为】:1  
151. 【现仓储量为】:1  
152. 【现仓储量为】:2  
153. 【现仓储量为】:3  
154. 【现仓储量为】:4  
155. 【现仓储量为】:0  
156. 【现仓储量为】:1  
157. 【现仓储量为】:5  
158. 【现仓储量为】:6  
159. 【现仓储量为】:7  
160. 【现仓储量为】:8  
161. 【现仓储量为】:9  
162. 【现仓储量为】:10  
163. 【现仓储量为】:11  
164. 【现仓储量为】:12  
165. 【现仓储量为】:13  
166. 【现仓储量为】:14  
167. 【现仓储量为】:15  
168. 【现仓储量为】:16  
169. 【现仓储量为】:17  
170. 【现仓储量为】:1  
171. 【现仓储量为】:1  
172. 【现仓储量为】:2  
173. 【现仓储量为】:3  
174. 【现仓储量为】:4  
175. 【现仓储量为】:5  
176. 【现仓储量为】:6  
177. 【现仓储量为】:3  
178. 【现仓储量为】:3  
179. 【现仓储量为】:1  
180. 【现仓储量为】:2  
181. 【现仓储量为】:3  
182. 【现仓储量为】:4  
183. 【现仓储量为】:5  
184. 【现仓储量为】:6  
185. 【现仓储量为】:7  
186. 【现仓储量为】:8  
187. 【现仓储量为】:9  
188. 【现仓储量为】:10  
189. 【现仓储量为】:11  
190. 【现仓储量为】:12  
191. 【现仓储量为】:13  
192. 【现仓储量为】:14  
193. 【现仓储量为】:15  
194. 【现仓储量为】:16  
195. 【现仓储量为】:17  
196. 【现仓储量为】:18  
197. 【现仓储量为】:19  
198. 【现仓储量为】:6  
199. 【现仓储量为】:7  
200. 【现仓储量为】:8  
201. 【现仓储量为】:9  
202. 【现仓储量为】:10  
203. 【现仓储量为】:11  
204. 【现仓储量为】:12  
205. 【现仓储量为】:13  
206. 【现仓储量为】:14  
207. 【现仓储量为】:15  
208. 【现仓储量为】:16  
209. 【现仓储量为】:17  
210. 【现仓储量为】:18  
211. 【现仓储量为】:19  
212. 【现仓储量为】:20  
213. 【现仓储量为】:21  
214. 【现仓储量为】:22  
215. 【现仓储量为】:23  
216. 【现仓储量为】:24  
217. 【现仓储量为】:25  
218. 【现仓储量为】:26  
219. 【现仓储量为】:27  
220. 【现仓储量为】:28  
221. 【现仓储量为】:29  
222. 【现仓储量为】:30  
223. 【现仓储量为】:31  
224. 【现仓储量为】:32  
225. 【现仓储量为】:33  
226. 【现仓储量为】:34  
227. 【现仓储量为】:35  
228. 【现仓储量为】:36  
229. 【现仓储量为】:37  
230. 【现仓储量为】:38  
231. 【现仓储量为】:39  
232. 【现仓储量为】: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你们能够放心使用，这可不是它的问题，只是在它和别的对象之间的同步有问题。