Producer Consumer using BlockingQueue

drainTo(Collection<? super E> c, int maxElements) Removes at most the given number of available elements from this queue and adds them to the given collection.

If you happen to use Google Guava, there's a nifty Queues.drain() method.

Drains the queue as BlockingQueue.drainTo(Collection, int), but if the requested numElements elements are not available, it will wait for them up to the specified timeout.

Producer Consumer using BlockingQueue - 我的博客 - ITeye技术网站

Usage example, based on a typical producer-consumer scenario. Note that a BlockingQueue can safely be used with multiple producers and multiple consumers.

1. class Producer implements Runnable {  
2.   private final BlockingQueue queue;  
3.   Producer(BlockingQueue q) { queue = q; }  
4.   public void run() {  
5.     try {  
6.       while (true) { queue.put(produce()); }  
7.     } catch (InterruptedException ex) { ... handle ...}  
8.   }  
9.   Object produce() { ... }  
10. }  
11.   
12. class Consumer implements Runnable {  
13.   private final BlockingQueue queue;  
14.   Consumer(BlockingQueue q) { queue = q; }  
15.   public void run() {  
16.     try {  
17.       while (true) { consume(queue.take()); }  
18.     } catch (InterruptedException ex) { ... handle ...}  
19.   }  
20.   void consume(Object x) { ... }  
21. }  

Memory consistency effects: As with other concurrent collections, actions in a thread prior to placing an object into a BlockingQueue happen-before actions subsequent to the access or removal of that element from theBlockingQueue in another thread.

BlockingQueue是Java Collection框架的一个接口，下面有很多实现类，包括：

ArrayBlockingQueue, (这个要自己会实现)

DelayQueue,  

LinkedBlockingQueue, 

PriorityBlockingQueue

不用BlockingQueue实现的话，可以这么做：

1. public class Drop {  
2.     // Message sent from producer  
3.     // to consumer.  
4.     private String message;  
5.     // True if consumer should wait  
6.     // for producer to send message,  
7.     // false if producer should wait for  
8.     // consumer to retrieve message.  
9.     private boolean empty = true;  
10.   
11.     public synchronized String take() {  
12.         // Wait until message is  
13.         // available.  
14.         while (empty) {  
15.             try {  
16.                 wait();  
17.             } catch (InterruptedException e) {}  
18.         }  
19.         // Toggle status.  
20.         empty = true;  
21.         // Notify producer that  
22.         // status has changed.  
23.         notifyAll();  
24.         return message;  
25.     }  
26.   
27.     public synchronized void put(String message) {  
28.         // Wait until message has  
29.         // been retrieved.  
30.         while (!empty) {  
31.             try {   
32.                 wait();  
33.             } catch (InterruptedException e) {}  
34.         }  
35.         // Toggle status.  
36.         empty = false;  
37.         // Store message.  
38.         this.message = message;  
39.         // Notify consumer that status  
40.         // has changed.  
41.         notifyAll();  
42.     }  
43. }  
44.   
45.   
46. public class Producer implements Runnable {  
47.     private Drop drop;  
48.   
49.     public Producer(Drop drop) {  
50.         this.drop = drop;  
51.     }  
52.   
53.     public void run() {  
54.         String importantInfo[] = {  
55.             "Mares eat oats",  
56.             "Does eat oats",  
57.             "Little lambs eat ivy",  
58.             "A kid will eat ivy too"  
59.         };  
60.         Random random = new Random();  
61.   
62.         for (int i = 0;  
63.              i < importantInfo.length;  
64.              i++) {  
65.             drop.put(importantInfo[i]);  
66.             try {  
67.                 Thread.sleep(random.nextInt(5000));  
68.             } catch (InterruptedException e) {}  
69.         }  
70.         drop.put("DONE");  
71.     }  
72. }  
73.   
74. public class Consumer implements Runnable {    
75.     private Drop drop;    
76.     
77.     public Consumer(Drop drop) {    
78.         this.drop = drop;    
79.     }    
80.     
81.     public void run() {    
82.         Random random = new Random();    
83.         for (String message = drop.take();    
84.              ! message.equals("DONE");    
85.              message = drop.take()) {    
86.             System.out.format("MESSAGE RECEIVED: %s%n", message);    
87.             try {    
88.                 Thread.sleep(random.nextInt(5000));    
89.             } catch (InterruptedException e) {}    
90.         }    
91.     }    
92. }

Read full article from Producer Consumer using BlockingQueue - 我的博客 - ITeye技术网站