BlockingQueue简介

BlockingQueue,根据其英文释义，为阻塞队列。根据队列的先进先出原则，可以得到其简单的示意图。



下面是BlockingQueue的主要方法。

对于插入操作，add、offer,put的三个方法的区别在于，offer方法比add更适用于空间有限制的queue,put方法则是用来等待当queue有空间的时候去插入。最后一个offer和put类似。但是可以用来指定等待时间。

class Producer implements Runnable {
private final BlockingQueue queue;
int productCount;
Producer(BlockingQueue q) { queue = q; }
public void run() {
try {
while (true) {

Thread.sleep(100);
Object product = produce();
boolean addResult;
/*
* put 方法，会在这里造成线程堵塞
*/
//queue.put(product);

/*
* add 方法会在容量不足的情况下抛出异常
*/
//addResult = queue.add(product);

/*
* offer 方法会在容量不足的情况下返回false来指示
*/
//addResult = queue.offer(product);
//System.out.println("add success ?" +addResult);
/*
* offer(e,timeout,unit) 方法会通过一个时间来等待
*/
addResult = queue.offer(product,800,TimeUnit.MILLISECONDS);
System.out.println("add success ?" +addResult);
System.out.println("produce: " + product+" at time:" + System.currentTimeMillis());
}
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
Object produce() {
return "product" + productCount++;
}
}
class Consumer implements Runnable {
private final BlockingQueue queue;

Consumer(BlockingQueue q) {
queue = q;
}

public void run() {
try {

while (true) {
Thread.sleep(100);
//只有当size大于等于10的时候才开始消费
if (queue.size() >= 10){
Thread.sleep(1000);
consume(queue.take());
}
}
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}

void consume(Object x) {
System.out.println("Thread :"+Thread.currentThread().getId() + "consume " + x );
System.out.println("time:" +System.currentTimeMillis());
}
}
public class BlockintQueueDemo {

public static void main(String[] args) {
LinkedBlockingQueue<String> q = new LinkedBlockingQueue<String>(10);
Producer p = new Producer(q);
Consumer c1 = new Consumer(q);
Consumer c2 = new Consumer(q);
Thread producerThread = new Thread(p);
Thread consumerThread1 = new Thread(c1);
Thread consumerThread2 = new Thread(c2);
producerThread.start();
consumerThread1.start();
consumerThread2.start();
}
}

使用add 方法的添加

先来看用put方法 添加 的打印信息

再看看 使用 offer方法的打印信息，或许你会纳闷为什么都超过阻塞队列容器的限度了还在一直生产

最后看一下offer(e,timeOut,unit)方法，在这里我设置了 800ms的等待时间，通过下面标示出的时间戳和添加结果可以看出，800ms显然太短，以为我们消费需要1000ms。所以最后的添加结果是false。另外，通过时间戳可以发现，在这里依然起了线程堵塞的作用。标示出的时间戳差距是 900ms。

二：通过offer 和 poll方法的实现的源码来比较 ArrayBlockingQueue 和 LinkedBlockingQueue的一些却别

通过英文释义可以知道ArrayBlockingQueue和LinkedBlockingQueue的区别之一就是存储数据的方式，前者是数组，后者是链表。

<span style="font-size:18px;">    public boolean offer(E e) {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lock();
try {
if (count == items.length)
return false;
else {
enqueue(e);
return true;
}
} finally {
lock.unlock();
}
}

public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return (count == 0) ? null : dequeue();
} finally {
lock.unlock();
}
}</span>

<span style="font-size:18px;">    public boolean offer(E e) {
if (e == null) throw new NullPointerException();
final AtomicInteger count = this.count;
if (count.get() == capacity)
return false;
int c = -1;
Node<E> node = new Node<E>(e);
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {
if (count.get() < capacity) {
enqueue(node);
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
}
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
return c >= 0;
}

public E poll(long timeout, TimeUnit unit) throws InterruptedException {
E x = null;
int c = -1;
long nanos = unit.toNanos(timeout);
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();
try {
while (count.get() == 0) {
if (nanos <= 0)
return null;
nanos = notEmpty.awaitNanos(nanos);
}
x = dequeue();
c = count.getAndDecrement();
if (c > 1)
notEmpty.signal();
} finally {
takeLock.unlock();
}
if (c == capacity)
signalNotFull();
return x;
}
</span>

1. ArrayBlockingQueue

基于数组的阻塞队列实现，在ArrayBlockingQueue内部，维护了一个定长数组，以便缓存队列中的数据对象，这是一个常用的阻塞队列，除了一个定长数组外，ArrayBlockingQueue内部还保存着两个整形变量，分别标识着队列的头部和尾部在数组中的位置。

　　ArrayBlockingQueue在生产者放入数据和消费者获取数据，都是共用同一个锁对象，由此也意味着两者无法真正并行运行，这点尤其不同于LinkedBlockingQueue；按照实现原理来分析，ArrayBlockingQueue完全可以采用分离锁，从而实现生产者和消费者操作的完全并行运行。Doug
Lea之所以没这样去做，也许是因为ArrayBlockingQueue的数据写入和获取操作已经足够轻巧，以至于引入独立的锁机制，除了给代码带来额外的复杂性外，其在性能上完全占不到任何便宜。 ArrayBlockingQueue和LinkedBlockingQueue间还有一个明显的不同之处在于，前者在插入或删除元素时不会产生或销毁任何额外的对象实例，而后者则会生成一个额外的Node对象。这在长时间内需要高效并发地处理大批量数据的系统中，其对于GC的影响还是存在一定的区别。而在创建ArrayBlockingQueue时，我们还可以控制对象的内部锁是否采用公平锁，默认采用非公平锁。

2. LinkedBlockingQueue

基于链表的阻塞队列，同ArrayListBlockingQueue类似，其内部也维持着一个数据缓冲队列（该队列由一个链表构成），当生产者往队列中放入一个数据时，队列会从生产者手中获取数据，并缓存在队列内部，而生产者立即返回；只有当队列缓冲区达到最大值缓存容量时（LinkedBlockingQueue可以通过构造函数指定该值），才会阻塞生产者队列，直到消费者从队列中消费掉一份数据，生产者线程会被唤醒，反之对于消费者这端的处理也基于同样的原理。而LinkedBlockingQueue之所以能够高效的处理并发数据，还因为其对于生产者端和消费者端分别采用了独立的锁来控制数据同步，这也意味着在高并发的情况下生产者和消费者可以并行地操作队列中的数据，以此来提高整个队列的并发性能。

作为开发者，我们需要注意的是，如果构造一个LinkedBlockingQueue对象，而没有指定其容量大小，LinkedBlockingQueue会默认一个类似无限大小的容量（Integer.MAX_VALUE），这样的话，如果生产者的速度一旦大于消费者的速度，也许还没有等到队列满阻塞产生，系统内存就有可能已被消耗殆尽了。

ArrayBlockingQueue和LinkedBlockingQueue是两个最普通也是最常用的阻塞队列，一般情况下，在处理多线程间的生产者消费者问题，使用这两个类足以。