BlockingQueue

package com.judy.queue;

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

public class BlockQueueT {

	public static void main(String[] args) throws Exception {
		// 声明一个容量为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);

	    // 借助Executors
	    ExecutorService service = Executors.newCachedThreadPool();
	    // 启动线程
	    service.execute(producer1);
	    service.execute(producer2);
	    service.execute(producer3);
	    service.execute(consumer);

	    // 执行4s
	    Thread.sleep(4 * 1000);
	    producer1.stop();
	    producer2.stop();
	    producer3.stop();

	    Thread.sleep(2000);
	    // 退出Executor
	    service.shutdown();
	}

}

package com.judy.queue;

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

public class Consumer implements Runnable {
	private BlockingQueue<String> queue;
	private static final int DEFAULT_RANGE_FOR_SLEEP = 1000;

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

	@Override
	public void run() {
		System.out.println("启动消费者线程启动...");
		Random r = new Random();
		boolean isRunning = true;
		try {
			while (isRunning) {
				System.out.println("正在往队列中存放数据...");
				String 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 (Exception e) {
			e.printStackTrace();
			Thread.currentThread().interrupt();
		} finally {
			System.out.println("退出消费者线程！");
		}
	}

}

package com.judy.queue;

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 BlockingQueue<String> queue;
	private volatile boolean isRunning = true;// 关于volatile,保证从主内存中取出的变量值是最新值(多个线程同时修改还是会出现非预想值的问题)
	// 一个提供原子操作的Integer的类。在Java语言中，++i和i++操作并不是线程安全的，在使用的时候，不可避免的会用到synchronized关键字。而AtomicInteger则通过一种线程安全的加减操作接口。
	private static AtomicInteger count= new AtomicInteger();
	private static final int DEFAULT_RANGE_FOR_SLEEP = 1000;

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

	@Override
	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 + "放入队列...");
				if (queue.offer(data, 2, TimeUnit.SECONDS)) {//在指定时间内将数据放入队列中
					System.out.println("放入数据失败：" + data);
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
			Thread.currentThread().interrupt();//要么中断标示会被清除  要么抛出异常
		} finally {
			System.out.println("退出生产者线程！");
		}

	}

	public void stop() {
		isRunning = false;
	}

}