使用BlockingQueue的生产者消费者模式
BlockingQueue很好的解决了多线程中，如何高效安全“传输”数据的问题。通过这些高效并且线程安全的队列类，为我们快速搭建高质量的多线程程序带来极大的便利。使用场景。
首先它是一个队列，而一个队列在数据结构中所起的作用大致如下图所示：
通过一个共享的队列，可以使得数据由队列的一端输入，从另外一端输出；在生产者消费者模式中，通过队列的方式可以很方便的实现两者之间的数据共享。强大的BlockingQueue使我们不用关心什么时候需要阻塞线程，什么时候需要唤醒线程。
BlockingQueue的核心方法：
放入数据：
offer(anObject) 如果BlockingQueue可以容纳，返回为true,否则返回false.
offer(E o,long timeout,TimeUnit unit),设置等待时间，如果指定时间内，还不能往队列中加入BlockingQueue,则返回失败。
put(anObject)把anObject加到BlockingQueue中，如果BlockQueue没有空间，则调用此方法的线程被阻断，直到BlockingQueue里面有空间再继续。
获取数据： poll(time):取走BlockingQueue里排在首位的对象,若不能立即取出,则可以等time参数规定的时间, 取不到时返回null; poll(long timeout, TimeUnit unit)：从BlockingQueue取出一个队首的对象，如果在指定时间内， 队列一旦有数据可取，则立即返回队列中的数据。否则知道时间超时还没有数据可取，返回失败。 take():取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到 BlockingQueue有新的数据被加入; drainTo():一次性从BlockingQueue获取所有可用的数据对象（还可以指定获取数据的个数）， 通过该方法，可以提升获取数据效率；不需要多次分批加锁或释放锁。
测试代码：
package BlockingQueue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class BlockingQueueTest {
public static void main(String args[]) throws InterruptedException{
BlockingQueue<String> queue = new ArrayBlockingQueue(10);
Producer producer1 = new Producer(queue);
Producer producer2 = new Producer(queue);
Producer producer3 = new Producer(queue);
Consumer consumer = new Consumer(queue);
ExecutorService service = Executors.newCachedThreadPool();
service.execute(producer1);
service.execute(producer2);
service.execute(producer3);
service.execute(consumer);
Thread.sleep(10 * 1000);
producer1.stop();
producer2.stop();
producer3.stop();
Thread.sleep(2000);
// 退出Executor
service.shutdown();
}
}生产者：
package BlockingQueue;
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<String> queue;
private static AtomicInteger count = new AtomicInteger();
private static final int DEFAULT_RANGE_FOR_SLEEP = 1000;
public Producer(BlockingQueue 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 + "放入队列...");
if (!queue.offer(data, 2, TimeUnit.SECONDS)) {
System.out.println("放入数据失败：" + data);
}
}
}catch (InterruptedException e) {
e.printStackTrace();
Thread.currentThread().interrupt();
}
finally{
System.out.println("退出生产者线程！");
}
}
public void stop(){
isRunning = false;
}
}消费者：
package BlockingQueue;
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;
}
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{
isRunning = false;
}
}
}catch(InterruptedException e){
e.printStackTrace();
Thread.currentThread().interrupt();
}finally{
System.out.println("退出消费者线程！");
}
}
}
参考：http://wsmajunfeng.iteye.com/blog/1629354
腾讯云开发者
