阻塞队列和生产者-消费者模式

 自Java5以来提供的BlockingQueue是一种特殊的队列, 它 是支持两个附加操作的 Queue，这两个操作是：检索元素时等待队列变为非空，以及存储元素时等待空间变得可用。

以JDK中的例子略加改写如下

1 import java.util.concurrent.ArrayBlockingQueue;
2 import java.util.concurrent.BlockingQueue;
3
4 class Producer implements Runnable {
5     private final BlockingQueue queue;
6
7      Producer(BlockingQueue q) {
8          queue = q;
9      }
10
11     public void run() {
12         try {
13             while (true) {
14                  queue.put(produce());
15              }
16          } catch (InterruptedException ex) {
17              System.out.println("produce interrupted " + ex.getMessage());
18              Thread.currentThread().interrupt();
19             //return;
20          }
21      }
22
23      Object produce() {
24          System.out.println("produce laugh");
25         return "haha";
26      }
27 }
28
29 class Consumer implements Runnable {
30     private final BlockingQueue queue;
31
32      Consumer(BlockingQueue q) {
33          queue = q;
34      }
35
36     public void run() {
37         try {
38             while (true) {
39                  consume(queue.take());
40              }
41          } catch (InterruptedException ex) {
42              System.out.println("consume interrupted " + ex.getMessage());
43              Thread.currentThread().interrupt();
44          }
45      }
46
47     void consume(Object x) {
48          System.out.println("consume laugh "+ x);
49      }
50 }
51
52 public class BlockingQueueTest {
53     public static void main(String args[]) {
54          BlockingQueue q = new ArrayBlockingQueue(10);
55          Producer p = new Producer(q);
56          Consumer c1 = new Consumer(q);
57          Consumer c2 = new Consumer(q);
58          Thread pTh = new Thread(p);
59          pTh.start();
60          Thread cTh1 = new Thread(c1);
61          cTh1.start();
62          Thread cTh2 = new Thread(c2);
63          cTh2.start();
64         try {
65              Thread.sleep(3000);
66          }catch (Exception e) {
67             // TODO: handle exception
68          }
69          pTh.interrupt();
70          cTh1.interrupt();
71          cTh2.interrupt();
72      }
73 }
74

posted on 2008-07-26 23:42