java队列Queue

在Java多线程应用中，队列的使用率很高，多数生产消费模型的首选数据结构就是队列。Java提供的线程安全的Queue可以分为阻塞队列和非阻塞队列，其中阻塞队列的典型例子是BlockingQueue，非阻塞队列的典型例子是ConcurrentLinkedQueue，在实际应用中要根据实际需要选用阻塞队列或者非阻塞队列。
注：什么叫线程安全？这个首先要明确。线程安全的类 ，指的是类内共享的全局变量的访问必须保证是不受多线程形式影响的。如果由于多线程的访问（比如修改、遍历、查看）而使这些变量结构被破坏或者针对这些变量操作的原子性被破坏，则这个类就不是线程安全的。

BlockingQueue

BlockingQueue，顾名思义，“阻塞队列”：可以提供阻塞功能的队列。
首先，看看BlockingQueue提供的常用方法：

----	可能报异常	返回布尔值	可能阻塞	设定等待时间
入队	add(e)	offer(e)	put(e)	offer(e, timeout, unit)
出队	remove()	poll()	take()	poll(timeout, unit)
查看	element()	peek()	无	无

强调

• add(e) remove() element() 方法不会阻塞线程。当不满足约束条件时，会抛出IllegalStateException 异常。例如：当队列被元素填满后，再调用add(e)，则会抛出异常。
• offer(e) poll() peek() 方法即不会阻塞线程，也不会抛出异常。例如：当队列被元素填满后，再调用offer(e)，则不会插入元素，函数返回false。
• 要想要实现阻塞功能，需要调用put(e) take() 方法。当不满足约束条件时，会阻塞线程。

以ArrayBlockingQueue类为例：

• 第一类方法

public boolean add(E e) {
if (offer(e))
return true;
else
throw new IllegalStateException("Queue full");//队列已满，抛异常
}

public E remove() {
E x = poll();
if (x != null)
return x;
else
throw new NoSuchElementException();//队列为空，抛异常
}

• 第二类方法

public boolean offer(E e) {
if (e == null)throw new NullPointerException();
final ReentrantLock lock = this.lock;
lock.lock();
try {
if (count == items.length)//队列已满，返回false
return false;
else {
insert(e);//insert方法中发出了notEmpty.signal();
return true;
}
} finally {
lock.unlock();
}
}

public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
if (count == 0)//队列为空，返回false
return null;
E x = extract();//extract方法中发出了notFull.signal();
return x;
} finally {
lock.unlock();
}
}

• 第三类方法(这里面涉及到Condition类，简要提一下)
  await方法指：造成当前线程在接到信号或被中断之前一直处于等待状态。
  signal方法指：唤醒一个等待线程。

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这个条件，这时当前线程被阻塞
notFull.await();
} catch (InterruptedException ie) {
notFull.signal(); //唤醒受notFull阻塞的当前线程
throw ie;
}
insert(e);
} finally {
lock.unlock();
}
}

public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
try {
while (count == 0)//如果队列为空，等待notEmpty这个条件，这时当前线程被阻塞
notEmpty.await();
} catch (InterruptedException ie) {
notEmpty.signal();//唤醒受notEmpty阻塞的当前线程
throw ie;
}
E x = extract();
return x;
} finally {
lock.unlock();
}
}

• 第四类方法就是指在有必要时等待指定时间，就不详细说了。

BlockingQueue接口的具体实现类

• ArrayBlockingQueue，其构造函数必须带一个int参数来指明其大小
• LinkedBlockingQueue，若其构造函数带一个规定大小的参数，生成的BlockingQueue有大小限制，若不带大小参数，所生成的BlockingQueue的大小由Integer.MAX_VALUE来决定
• PriorityBlockingQueue，其所含对象的排序不是FIFO,而是依据对象的自然排序顺序或者是构造函数的Comparator决定的顺序

上面是用ArrayBlockingQueue举得例子，下面看看LinkedBlockingQueue：

首先，既然是链表，就应该有Node节点，它是一个内部静态类：

static class Node<E> {
/** The item, volatile to ensure barrier separating write and read */
volatile E item;
Node<E> next;
Node(E x) { item = x; }
}

然后，对于链表来说，肯定需要两个变量来标示头和尾：

/** 头指针 */
private transient Node<E> head;//head.next是队列的头元素
/** 尾指针 */
private transient Node<E> last;//last.next是null

那么，对于入队和出队就很自然能理解了：

private void enqueue(E x) {
last = last.next = new Node<E>(x);//入队是为last再找个下家
}

private E dequeue() {
Node<E> first = head.next; //出队是把head.next取出来，然后将head向后移一位
head = first;
E x = first.item;
first.item = null;
return x;
}

另外，LinkedBlockingQueue相对于ArrayBlockingQueue还有不同是，有两个ReentrantLock，且队列现有元素的大小由一个AtomicInteger对象标示。

• 注：AtomicInteger类是以原子的方式操作整型变量。

private final AtomicInteger count =new AtomicInteger(0);
/** 用于读取的独占锁*/
private final ReentrantLock takeLock =new ReentrantLock();
/** 队列是否为空的条件 */
private final Condition notEmpty = takeLock.newCondition();
/** 用于写入的独占锁 */
private final ReentrantLock putLock =new ReentrantLock();
/** 队列是否已满的条件 */
private final Condition notFull = putLock.newCondition();

有两个Condition很好理解，在ArrayBlockingQueue也是这样做的。但是为什么需要两个ReentrantLock呢？下面会慢慢道来。
让我们来看看offer和poll方法的代码：

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;
final ReentrantLock putLock =this.putLock;//入队当然用putLock
putLock.lock();
try {
if (count.get() < capacity) {
enqueue(e); //入队
c = count.getAndIncrement(); //队长度+1
if (c + 1 < capacity)
notFull.signal(); //队列没满，当然可以解锁了
}
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();//这个方法里发出了notEmpty.signal();
return c >= 0;
}

public E poll() {
final AtomicInteger count = this.count;
if (count.get() == 0)
return null;
E x = null;
int c = -1;
final ReentrantLock takeLock =this.takeLock;出队当然用takeLock
takeLock.lock();
try {
if (count.get() > 0) {
x = dequeue();//出队
c = count.getAndDecrement();//队长度-1
if (c > 1)
notEmpty.signal();//队列没空，解锁
}
} finally {
takeLock.unlock();
}
if (c == capacity)
signalNotFull();//这个方法里发出了notFull.signal();
return x;
}

看看源代码发现和上面ArrayBlockingQueue的很类似，关键的问题在于：为什么要用两个ReentrantLockputLock和takeLock？
我们仔细想一下，入队操作其实操作的只有队尾引用last，并且没有牵涉到head。而出队操作其实只针对head，和last没有关系。那么就是说入队和出队的操作完全不需要公用一把锁，所以就设计了两个锁，这样就实现了多个不同任务的线程入队的同时可以进行出队的操作，另一方面由于两个操作所共同使用的count是AtomicInteger类型的，所以完全不用考虑计数器递增递减的问题。
另外，还有一点需要说明一下：await()和singal()这两个方法执行时都会检查当前线程是否是独占锁的当前线程，如果不是则抛出java.lang.IllegalMonitorStateException异常。所以可以看到在源码中这两个方法都出现在Lock的保护块中。

=============分隔================

下面再来说说ConcurrentLinkedQueue，它是一个无锁的并发线程安全的队列。
对比锁机制的实现，使用无锁机制的难点在于要充分考虑线程间的协调。简单的说就是多个线程对内部数据结构进行访问时，如果其中一个线程执行的中途因为一些原因出现故障，其他的线程能够检测并帮助完成剩下的操作。这就需要把对数据结构的操作过程精细的划分成多个状态或阶段，考虑每个阶段或状态多线程访问会出现的情况。
ConcurrentLinkedQueue有两个volatile的线程共享变量：head，tail。要保证这个队列的线程安全就是保证对这两个Node的引用的访问（更新，查看）的原子性和可见性，由于volatile本身能够保证可见性，所以就是对其修改的原子性要被保证。
下面通过offer方法的实现来看看在无锁情况下如何保证原子性：

public boolean offer(E e) {
if (e == null)throw new NullPointerException();
Node<E> n = new Node<E>(e, null);
for (;;) {
Node<E> t = tail;
Node<E> s = t.getNext();
if (t == tail) { //------------------------------a
if (s == null) {//---------------------------b
if (t.casNext(s, n)) { //-------------------c
casTail(t, n); //------------------------d
return true;
}
} else {
casTail(t, s); //----------------------------e
}
}
}
}

此方法的循环内首先获得尾指针和其next指向的对象，由于tail和Node的next均是volatile的，所以保证了获得的分别都是最新的值。
代码a：t==tail是最上层的协调，如果其他线程改变了tail的引用，则说明现在获得不是最新的尾指针需要重新循环获得最新的值。
代码b：s==null的判断。静止状态下tail的next一定是指向null的，但是多线程下的另一个状态就是中间态：tail的指向没有改变，但是其next已经指向新的结点，即完成tail引用改变前的状态，这时候s!=null。这里就是协调的典型应用，直接进入代码e去协调参与中间态的线程去完成最后的更新，然后重新循环获得新的tail开始自己的新一次的入队尝试。另外值得注意的是a,b之间，其他的线程可能会改变tail的指向，使得协调的操作失败。从这个步骤可以看到无锁实现的复杂性。
代码c：t.casNext(s, n)是入队的第一步，因为入队需要两步：更新Node的next，改变tail的指向。代码c之前可能发生tail引用指向的改变或者进入更新的中间态，这两种情况均会使得t指向的元素的next属性被原子的改变，不再指向null。这时代码c操作失败，重新进入循环。
代码d：这是完成更新的最后一步了，就是更新tail的指向，最有意思的协调在这儿又有了体现。从代码看casTail(t, n)不管是否成功都会接着返回true标志着更新的成功。首先如果成功则表明本线程完成了两步的更新，返回true是理所当然的；如果 casTail(t, n)不成功呢？要清楚的是完成代码c则代表着更新进入了中间态，代码d不成功则是tail的指向被其他线程改变。意味着对于其他的线程而言：它们得到的是中间态的更新，s!=null，进入代码e帮助本线程执行最后一步并且先于本线程成功。这样本线程虽然代码d失败了，但是是由于别的线程的协助先完成了，所以返回true也就理所当然了。
通过分析这个入队的操作，可以清晰的看到无锁实现的每个步骤和状态下多线程之间的协调和工作。
注：上面这大段文字看起来很累，先能看懂多少看懂多少，现在看不懂先不急，下面还会提到这个算法，并且用示意图说明，就易懂很多了。

在使用ConcurrentLinkedQueue时要注意，如果直接使用它提供的函数，比如add或者poll方法，这样我们自己不需要做任何同步。
但如果是非原子操作，比如：

if(!queue.isEmpty()) {
queue.poll(obj);
}

我们很难保证，在调用了isEmpty()之后，poll()之前，这个queue没有被其他线程修改。所以对于这种情况，我们还是需要自己同步：

synchronized(queue) {
if(!queue.isEmpty()) {
queue.poll(obj);
}
}

• 注：这种需要进行自己同步的情况要视情况而定，不是任何情况下都需要这样做。
  另外还说一下，ConcurrentLinkedQueue的size()是要遍历一遍集合的，所以尽量要避免用size而改用isEmpty()，以免性能过慢。

[文章转载自]madun大神