深入理解Java线程编程中的阻塞队列容器

1. 什么是阻塞队列？

阻塞队列（blockingqueue）是一个支持两个附加操作的队列。这两个附加的操作是：在队列为空时，获取元素的线程会等待队列变为非空。当队列满时，存储元素的线程会等待队列可用。阻塞队列常用于生产者和消费者的场景，生产者是往队列里添加元素的线程，消费者是从队列里拿元素的线程。阻塞队列就是生产者存放元素的容器，而消费者也只从容器里拿元素。

阻塞队列提供了四种处理方法:

抛出异常：是指当阻塞队列满时候，再往队列里插入元素，会抛出illegalstateexception("queue full")异常。当队列为空时，从队列里获取元素时会抛出nosuchelementexception异常 。
返回特殊值：插入方法会返回是否成功，成功则返回true。移除方法，则是从队列里拿出一个元素，如果没有则返回null
一直阻塞：当阻塞队列满时，如果生产者线程往队列里put元素，队列会一直阻塞生产者线程，直到拿到数据，或者响应中断退出。当队列空时，消费者线程试图从队列里take元素，队列也会阻塞消费者线程，直到队列可用。
超时退出：当阻塞队列满时，队列会阻塞生产者线程一段时间，如果超过一定的时间，生产者线程就会退出。
2. java里的阻塞队列

jdk7提供了7个阻塞队列。分别是

1. arrayblockingqueue ：一个由数组结构组成的有界阻塞队列。
2. linkedblockingqueue ：一个由链表结构组成的有界阻塞队列。
3. priorityblockingqueue ：一个支持优先级排序的*阻塞队列。
4. delayqueue：一个使用优先级队列实现的*阻塞队列。
5. synchronousqueue：一个不存储元素的阻塞队列。
6. linkedtransferqueue：一个由链表结构组成的*阻塞队列。
7. linkedblockingdeque：一个由链表结构组成的双向阻塞队列。

arrayblockingqueue是一个用数组实现的有界阻塞队列。此队列按照先进先出（fifo）的原则对元素进行排序。默认情况下不保证访问者公平的访问队列，所谓公平访问队列是指阻塞的所有生产者线程或消费者线程，当队列可用时，可以按照阻塞的先后顺序访问队列，即先阻塞的生产者线程，可以先往队列里插入元素，先阻塞的消费者线程，可以先从队列里获取元素。通常情况下为了保证公平性会降低吞吐量。我们可以使用以下代码创建一个公平的阻塞队列：

arrayblockingqueue fairqueue = new arrayblockingqueue(1000,true);

访问者的公平性是使用可重入锁实现的，代码如下：

public arrayblockingqueue(int capacity, boolean fair) {
    if (capacity <= 0)
      throw new illegalargumentexception();
    this.items = new object[capacity];
    lock = new reentrantlock(fair);
    notempty = lock.newcondition();
    notfull = lock.newcondition();
}

linkedblockingqueue是一个用链表实现的有界阻塞队列。此队列的默认和最大长度为integer.max_value。此队列按照先进先出的原则对元素进行排序。

priorityblockingqueue是一个支持优先级的*队列。默认情况下元素采取自然顺序排列，也可以通过比较器comparator来指定元素的排序规则。元素按照升序排列。

delayqueue是一个支持延时获取元素的*阻塞队列。队列使用priorityqueue来实现。队列中的元素必须实现delayed接口，在创建元素时可以指定多久才能从队列中获取当前元素。只有在延迟期满时才能从队列中提取元素。我们可以将delayqueue运用在以下应用场景：

缓存系统的设计：可以用delayqueue保存缓存元素的有效期，使用一个线程循环查询delayqueue，一旦能从delayqueue中获取元素时，表示缓存有效期到了。
定时任务调度。使用delayqueue保存当天将会执行的任务和执行时间，一旦从delayqueue中获取到任务就开始执行，从比如timerqueue就是使用delayqueue实现的。
队列中的delayed必须实现compareto来指定元素的顺序。比如让延时时间最长的放在队列的末尾。实现代码如下：

public int compareto(delayed other) {
      if (other == this) // compare zero only if same object
        return 0;
      if (other instanceof scheduledfuturetask) {
        scheduledfuturetask x = (scheduledfuturetask)other;
        long diff = time - x.time;
        if (diff < 0)
          return -1;
        else if (diff > 0)
          return 1;
  else if (sequencenumber < x.sequencenumber)
          return -1;
        else
          return 1;
      }
      long d = (getdelay(timeunit.nanoseconds) -
           other.getdelay(timeunit.nanoseconds));
      return (d == 0) ? 0 : ((d < 0) ? -1 : 1);
    }

3.如何实现delayed接口

我们可以参考scheduledthreadpoolexecutor里scheduledfuturetask类。这个类实现了delayed接口。首先：在对象创建的时候，使用time记录前对象什么时候可以使用，代码如下：

scheduledfuturetask(runnable r, v result, long ns, long period) {
      super(r, result);
      this.time = ns;
      this.period = period;
      this.sequencenumber = sequencer.getandincrement();
}

然后使用getdelay可以查询当前元素还需要延时多久，代码如下：

public long getdelay(timeunit unit) {
      return unit.convert(time - now(), timeunit.nanoseconds);
    }

通过构造函数可以看出延迟时间参数ns的单位是纳秒，自己设计的时候最好使用纳秒，因为getdelay时可以指定任意单位，一旦以纳秒作为单位，而延时的时间又精确不到纳秒就麻烦了。使用时请注意当time小于当前时间时，getdelay会返回负数。

4.如何实现延时队列

延时队列的实现很简单，当消费者从队列里获取元素时，如果元素没有达到延时时间，就阻塞当前线程。

long delay = first.getdelay(timeunit.nanoseconds);
          if (delay <= 0)
            return q.poll();
          else if (leader != null)
            available.await();

synchronousqueue是一个不存储元素的阻塞队列。每一个put操作必须等待一个take操作，否则不能继续添加元素。synchronousqueue可以看成是一个传球手，负责把生产者线程处理的数据直接传递给消费者线程。队列本身并不存储任何元素，非常适合于传递性场景,比如在一个线程中使用的数据，传递给另外一个线程使用，synchronousqueue的吞吐量高于linkedblockingqueue 和 arrayblockingqueue。

linkedtransferqueue是一个由链表结构组成的*阻塞transferqueue队列。相对于其他阻塞队列，linkedtransferqueue多了trytransfer和transfer方法。

transfer方法。如果当前有消费者正在等待接收元素（消费者使用take()方法或带时间限制的poll()方法时），transfer方法可以把生产者传入的元素立刻transfer（传输）给消费者。如果没有消费者在等待接收元素，transfer方法会将元素存放在队列的tail节点，并等到该元素被消费者消费了才返回。transfer方法的关键代码如下：

node pred = tryappend(s, havedata);
return awaitmatch(s, pred, e, (how == timed), nanos);

第一行代码是试图把存放当前元素的s节点作为tail节点。第二行代码是让cpu自旋等待消费者消费元素。因为自旋会消耗cpu，所以自旋一定的次数后使用thread.yield()方法来暂停当前正在执行的线程，并执行其他线程。

trytransfer方法。则是用来试探下生产者传入的元素是否能直接传给消费者。如果没有消费者等待接收元素，则返回false。和transfer方法的区别是trytransfer方法无论消费者是否接收，方法立即返回。而transfer方法是必须等到消费者消费了才返回。

对于带有时间限制的trytransfer(e e, long timeout, timeunit unit)方法，则是试图把生产者传入的元素直接传给消费者，但是如果没有消费者消费该元素则等待指定的时间再返回，如果超时还没消费元素，则返回false，如果在超时时间内消费了元素，则返回true。

linkedblockingdeque是一个由链表结构组成的双向阻塞队列。所谓双向队列指的你可以从队列的两端插入和移出元素。双端队列因为多了一个操作队列的入口，在多线程同时入队时，也就减少了一半的竞争。相比其他的阻塞队列，linkedblockingdeque多了addfirst，addlast，offerfirst，offerlast，peekfirst，peeklast等方法，以first单词结尾的方法，表示插入，获取（peek）或移除双端队列的第一个元素。以last单词结尾的方法，表示插入，获取或移除双端队列的最后一个元素。另外插入方法add等同于addlast，移除方法remove等效于removefirst。但是take方法却等同于takefirst，不知道是不是jdk的bug，使用时还是用带有first和last后缀的方法更清楚。

在初始化linkedblockingdeque时可以设置容量防止其过渡膨胀。另外双向阻塞队列可以运用在“工作窃取”模式中。

5.阻塞队列的实现原理
本文以arrayblockingqueue为例，其他阻塞队列实现原理可能和arrayblockingqueue有一些差别，但是大体思路应该类似，有兴趣的朋友可自行查看其他阻塞队列的实现源码。

　　首先看一下arrayblockingqueue类中的几个成员变量：

public class arrayblockingqueue<e> extends abstractqueue<e>
implements blockingqueue<e>, java.io.serializable {
 
private static final long serialversionuid = -817911632652898426l;
 
/** the queued items */
private final e[] items;
/** items index for next take, poll or remove */
private int takeindex;
/** items index for next put, offer, or add. */
private int putindex;
/** number of items in the queue */
private int count;
 
/*
* concurrency control uses the classic two-condition algorithm
* found in any textbook.
*/
 
/** main lock guarding all access */
private final reentrantlock lock;
/** condition for waiting takes */
private final condition notempty;
/** condition for waiting puts */
private final condition notfull;
}

 　　可以看出，arrayblockingqueue中用来存储元素的实际上是一个数组，takeindex和putindex分别表示队首元素和队尾元素的下标，count表示队列中元素的个数。

　　lock是一个可重入锁，notempty和notfull是等待条件。

　　下面看一下arrayblockingqueue的构造器，构造器有三个重载版本：

public arrayblockingqueue(int capacity) {
}
public arrayblockingqueue(int capacity, boolean fair) {
 
}
public arrayblockingqueue(int capacity, boolean fair,
             collection<? extends e> c) {
}

 　　第一个构造器只有一个参数用来指定容量，第二个构造器可以指定容量和公平性，第三个构造器可以指定容量、公平性以及用另外一个集合进行初始化。

　　然后看它的两个关键方法的实现：put()和take()：

public void put(e e) throws interruptedexception {
  if (e == null) throw new nullpointerexception();
  final e[] items = this.items;
  final reentrantlock lock = this.lock;
  lock.lockinterruptibly();
  try {
    try {
      while (count == items.length)
        notfull.await();
    } catch (interruptedexception ie) {
      notfull.signal(); // propagate to non-interrupted thread
      throw ie;
    }
    insert(e);
  } finally {
    lock.unlock();
  }
}

 　　从put方法的实现可以看出，它先获取了锁，并且获取的是可中断锁，然后判断当前元素个数是否等于数组的长度，如果相等，则调用notfull.await()进行等待，如果捕获到中断异常，则唤醒线程并抛出异常。

　　当被其他线程唤醒时，通过insert(e)方法插入元素，最后解锁。

　　我们看一下insert方法的实现：

private void insert(e x) {
  items[putindex] = x;
  putindex = inc(putindex);
  ++count;
  notempty.signal();
}

 　　它是一个private方法，插入成功后，通过notempty唤醒正在等待取元素的线程。

　　下面是take()方法的实现：

public e take() throws interruptedexception {
  final reentrantlock lock = this.lock;
  lock.lockinterruptibly();
  try {
    try {
      while (count == 0)
        notempty.await();
    } catch (interruptedexception ie) {
      notempty.signal(); // propagate to non-interrupted thread
      throw ie;
    }
    e x = extract();
    return x;
  } finally {
    lock.unlock();
  }
}

 　　跟put方法实现很类似，只不过put方法等待的是notfull信号，而take方法等待的是notempty信号。在take方法中，如果可以取元素，则通过extract方法取得元素，下面是extract方法的实现：

private e extract() {
  final e[] items = this.items;
  e x = items[takeindex];
  items[takeindex] = null;
  takeindex = inc(takeindex);
  --count;
  notfull.signal();
  return x;
}

 　　跟insert方法也很类似。

　　其实从这里大家应该明白了阻塞队列的实现原理，事实它和我们用object.wait()、object.notify()和非阻塞队列实现生产者-消费者的思路类似，只不过它把这些工作一起集成到了阻塞队列中实现。