import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;/*** @ClassName Producer* @Description TODO(生产者)* @author wwj* @Date 2016年7月14日 下午4:22:57* @version 1.0.0.0*/
public class Producer implements Runnable {private volatile boolean      isRunning               = true;private BlockingQueue<String>  queue;private static AtomicInteger  count                   = new AtomicInteger();private static final int      DEFAULT_RANGE_FOR_SLEEP = 1000;public Producer(BlockingQueue<String> queue) {this.queue = queue;}public void run() {String data = null;Random r = new Random();System.out.println("启动生产者线程！");try {while (isRunning) {System.out.println("正在生产数据...");Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));data = "data:" + count.incrementAndGet();System.out.println("将数据：" + data + "放入队列...");//add(anObject):把anObject加到BlockingQueue里,即如果BlockingQueue可以容纳,则返回true,否则招聘异常//put(anObject):把anObject加到BlockingQueue里,如果BlockQueue没有空间,则调用此方法的线程被阻断直到BlockingQueue里面有空间再继续.//offer(anObject):表示如果可能的话,将anObject加到BlockingQueue里,即如果BlockingQueue可以容纳,则返回true,否则返回falseif (!queue.offer(data, 2L, TimeUnit.SECONDS)) {System.out.println("放入数据失败：" + data);}}} catch (InterruptedException e) {e.printStackTrace();Thread.currentThread().interrupt();} finally {System.out.println("退出生产者线程！");}}public void stop() {isRunning = false;}}

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;/*** @ClassName Consumer* @Description TODO(消费者)* @author wwj* @Date 2016年7月14日 下午4:22:23* @version 1.0.0.0*/
public class Consumer implements Runnable {public Consumer(BlockingQueue<String> queue) {this.queue = queue;}public void run() {System.out.println("启动消费者线程！");Random r = new Random();boolean isRunning = true;try {while (isRunning) {System.out.println("正从队列获取数据...");//take():取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到Blocking有新的对象被加入为止                 //如果用take -->   if (queue.size() > 0) {}//poll(time):取走BlockingQueue里排在首位的对象,若不能立即取出,则可以等time参数规定的时间,取不到时返回nullString data = queue.poll(2, TimeUnit.SECONDS);if (null != data) {System.out.println("拿到数据：" + data);System.out.println("正在消费数据：" + data);Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));} else {// 超过2s还没数据，认为所有生产线程都已经退出，自动退出消费线程。isRunning = false;}}} catch (InterruptedException e) {e.printStackTrace();Thread.currentThread().interrupt();} finally {System.out.println("退出消费者线程！");}}private BlockingQueue<String> queue;private static final int      DEFAULT_RANGE_FOR_SLEEP = 1000;
}

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;public class BlockingQueueTest {public static void main(String[] args) throws InterruptedException {// 声明一个容量为10的缓存队列BlockingQueue<String> queue = new LinkedBlockingQueue<String>(10);Producer producer1 = new Producer(queue);Producer producer2 = new Producer(queue);Producer producer3 = new Producer(queue);Consumer consumer = new Consumer(queue);// 借助ExecutorsExecutorService service = Executors.newCachedThreadPool();// 启动线程
        service.execute(producer1);service.execute(producer2);service.execute(producer3);service.execute(consumer);// 执行10sThread.sleep(10 * 1000);producer1.stop();producer2.stop();producer3.stop();Thread.sleep(2000);// 退出Executor
        service.shutdown();}
}