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

自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