BlockingQueue队列生产者消费者示例

1package org.web.controller.queue;
2
3import java.util.Random;
4import java.util.concurrent.BlockingQueue;
5import java.util.concurrent.TimeUnit;
6import java.util.concurrent.atomic.AtomicInteger;
7
8/**
9 * 生产者线程
10 */
11public class Producer implements Runnable {
12
13	public Producer(BlockingQueue queue) {
14		this.queue = queue;
15	}
16
17	public void run() {
18		String data = null;
19		Random r = new Random();
20
21		System.out.println("启动生产者线程！");
22		try {
23			while (isRunning) {
24				System.out.println("正在生产数据...");
25				Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));
26
27				data = "data:" + count.incrementAndGet();
28				System.out.println("将数据：" + data + "放入队列...");
29				if (!queue.offer(data, 2, TimeUnit.SECONDS)) {
30					System.out.println("放入数据失败：" + data);
31				}
32			}
33		} catch (InterruptedException e) {
34			e.printStackTrace();
35			Thread.currentThread().interrupt();
36		} finally {
37			System.out.println("退出生产者线程！");
38		}
39	}
40
41	public void stop() {
42		isRunning = false;
43	}
44
45	private volatile boolean isRunning = true;
46	private BlockingQueue queue;
47	private static AtomicInteger count = new AtomicInteger();
48	private static final int DEFAULT_RANGE_FOR_SLEEP = 1000;
49
50}
51
52
53package org.web.controller.queue;
54
55import java.util.Random;
56import java.util.concurrent.BlockingQueue;
57import java.util.concurrent.TimeUnit;
58
59/**
60 * 消费者线程
61 * 
62 */
63public class Consumer implements Runnable {
64
65	public Consumer(BlockingQueue<String> queue) {
66		this.queue = queue;
67	}
68
69	public void run() {
70		System.out.println("启动消费者线程！");
71		Random r = new Random();
72		boolean isRunning = true;
73		try {
74			while (isRunning) {
75				System.out.println("正从队列获取数据...");
76				String data = queue.poll(2, TimeUnit.SECONDS);
77				if (null != data) {
78					System.out.println("拿到数据：" + data);
79					System.out.println("正在消费数据：" + data);
80					Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));
81				} else {
82					// 超过2s还没数据，认为所有生产线程都已经退出，自动退出消费线程。
83					isRunning = false;
84				}
85			}
86		} catch (InterruptedException e) {
87			e.printStackTrace();
88			Thread.currentThread().interrupt();
89		} finally {
90			System.out.println("退出消费者线程！");
91		}
92	}
93
94	private BlockingQueue<String> queue;
95	private static final int DEFAULT_RANGE_FOR_SLEEP = 1000;
96}
97
98
99
100package org.web.controller.queue;
101
102import java.util.concurrent.BlockingQueue;
103import java.util.concurrent.ExecutorService;
104import java.util.concurrent.Executors;
105import java.util.concurrent.LinkedBlockingQueue;
106
107public class BlockingQueueTest {
108
109	public static void main(String[] args) throws InterruptedException {
110		// 声明一个容量为10的缓存队列
111		BlockingQueue<String> queue = new LinkedBlockingQueue<String>(10);
112
113		Producer producer1 = new Producer(queue);
114		Producer producer2 = new Producer(queue);
115		Producer producer3 = new Producer(queue);
116		Consumer consumer = new Consumer(queue);
117
118		// 借助Executors
119		ExecutorService service = Executors.newCachedThreadPool();
120		// 启动线程
121		service.execute(producer1);
122		service.execute(producer2);
123		service.execute(producer3);
124		service.execute(consumer);
125
126		// 执行10s
127		Thread.sleep(10 * 1000);
128		producer1.stop();
129		producer2.stop();
130		producer3.stop();
131
132		Thread.sleep(2000);
133		// 退出Executor
134		service.shutdown();
135	}
136}