首先建立生產者：

package ProducerAndConsumer;

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class Producer implements Runnable {

	private volatile boolean isRunning = true;
	private BlockingQueue<PCData> queue; // 記憶體緩衝區
	private static AtomicInteger count = new AtomicInteger();
	// 總數，原子操作
	private static final int SLEEPTIME = 1000;

	public Producer(BlockingQueue<PCData> queue) {
		this.queue = queue;
	}

	@Override
	public void run() {

		PCData data = null;
		Random r = new Random();
		System.out.println("start producer id=" + Thread.currentThread().getId());
		try {
			while (isRunning) {
				Thread.sleep(r.nextInt(SLEEPTIME));
				data = new PCData(count.incrementAndGet());// 構造任務資料
				System.out.println(data + "is put into queue");
				if (!queue.offer(data, 2, TimeUnit.SECONDS)) {
					// 提交資料到緩衝區中
					System.err.println("failed to put data:" + data);
				}
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	public void stop() {
		isRunning = false;
	}

}
 

package ProducerAndConsumer;


public class PCData {


<span style="white-space:pre">	</span>private final int intData;


<span style="white-space:pre">	</span>public PCData(int d) {
<span style="white-space:pre">		</span>intData = d;
<span style="white-space:pre">	</span>}


<span style="white-space:pre">	</span>public PCData(String d) {
<span style="white-space:pre">		</span>intData = Integer.valueOf(d);
<span style="white-space:pre">	</span>}


<span style="white-space:pre">	</span>public int getData() {
<span style="white-space:pre">		</span>return this.intData;
<span style="white-space:pre">	</span>}


<span style="white-space:pre">	</span>public String toString() {
<span style="white-space:pre">		</span>return "data:" + intData;
<span style="white-space:pre">	</span>}
}
 

package ProducerAndConsumer;

import java.text.MessageFormat;
import java.util.Random;
import java.util.concurrent.BlockingQueue;

public class Consumer implements Runnable {

	private BlockingQueue<PCData> queue;// 緩衝區
	
	private static final int SLEEPTIME = 1000;

	public Consumer(BlockingQueue<PCData> queue) {
		this.queue = queue;
	}

	@Override
	public void run() {

		System.out.println("start Consumer id=" + Thread.currentThread().getId());
		Random r = new Random(); // 隨機等待時間
		try {

			while (true) {
				PCData data = queue.take();// 提取任務
				if (null != data) {
					int re = data.getData() * data.getData();// 計算平方
					System.out.println(MessageFormat.format("{0}*{1}={2}",
							data.getData(), data.getData(), re));
					Thread.sleep(r.nextInt(SLEEPTIME));
				}
			}

		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}
 

package ProducerAndConsumer;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

public class Main {

	public static void main(String[] args) {

		try {
			// 建立緩衝區
			BlockingQueue<PCData> queue = new LinkedBlockingQueue<PCData>(10);
			Producer producer1 = new Producer(queue);// 建立生產者
			Producer producer2 = new Producer(queue);
			Producer producer3 = new Producer(queue);

			Consumer consumer1 = new Consumer(queue);// 建立消費者
			Consumer consumer2 = new Consumer(queue);
			Consumer consumer3 = new Consumer(queue);

			ExecutorService service = Executors.newCachedThreadPool();// 建立執行緒池

			service.execute(producer1);// 執行生產者
			service.execute(producer2);
			service.execute(producer3);

			service.execute(consumer1);// 執行消費者
			service.execute(consumer2);
			service.execute(consumer3);

			Thread.sleep(10 * 1000);

			producer1.stop();// 停止生產者
			producer2.stop();
			producer3.stop();

			Thread.sleep(3000);

			service.shutdown();

		} catch (InterruptedException e) {
			e.printStackTrace();
		}

	}
}