更多并发相关内容，查看==>Java 线程&并发学习目录

阻塞队列BlockingQueue是一种在多线程环境下依旧可以保证线程安全的FIFO队列，生产者从入口端添加数据消费者从出口段获取数据，而当队列为空时，消费者就被阻塞了，当队列无有效空间时，生产者就被阻塞了，这也是阻塞队列名字的由来，通过阻塞队列可以很高效的实现消费者和生产者模型的相关功能了。

在之前的学习笔记Java 利用wait和notify实现阻塞队列中已经介绍了通过Object的wait和notify方法实现一个消费者生产者模型的，而阻塞队列是如何在多线程的环境下保证消费者和生产者的有序操作的呢？这一切都是依靠着并发包中的条件队列Ccondition实现的，而条件队列又是依靠并发包中的核心组件AQS的。先看一个Condition的demo初探究竟。

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

class BoundedBuffer {
    final Lock lock = new ReentrantLock();
    final Condition notFull = lock.newCondition();
    final Condition notEmpty = lock.newCondition();
    // 通过同一个Lock生产的不同的condition对象

    final Object[] items = new Object[100];
    // 定长数组，生产者的数据会添加到此，消费者也会从该数组获取数据
    int putptr, takeptr, count;
    // 消费者和生产者的数组偏移量以及当前数组容量

    // 生产
    public void put(Object x) throws InterruptedException {
        lock.lock();
        try {
            while (count == items.length)
                notFull.await();  
                // 队列已满，生产者进入到等待状态，直到 not full 才能继续生产
            items[putptr] = x;
            if (++putptr == items.length) putptr = 0;
            ++count;
            notEmpty.signal(); 
            // 成功生成一条数据，通知消费者
        } finally {
            lock.unlock();
        }
    }

    // 消费
    public Object take() throws InterruptedException {
        lock.lock();
        try {
            while (count == 0)
                notEmpty.await(); // 队列为空，消费者进入到等待状态，直到队列 not empty，才能继续消费
            Object x = items[takeptr];
            if (++takeptr == items.length) takeptr = 0;
            --count;
            notFull.signal(); // 消费了一条数据，通知生产者
            return x;
        } finally {
            lock.unlock();
        }
    }
}

这其实是并发包的作者Doug Lea写的一个非常经典的消费者生产者模型代码。如果仔细看大部分代码和使用wait、notify没有太多的区别，只是改用了Condition，搭配着ReentrantLock使用的，在singal和await方法前必须使用ReentrantLock.lock 方法获取到当前的锁。接下来就来学习下Condition 条件队列的工作原理

初识 Condition 对象

lock.newCondition() 这句话就是生成一个Condition 对象

public Condition newCondition() {
    // sync对象是在ReentranctLock对象生成的Sync对象,有公平模式和非公平模式
    return sync.newCondition();
}

final ConditionObject newCondition() {
    return new ConditionObject();
}

ConditionObject 对象则包含了两个Node节点，代表了链表头部firstWaiter以及链表尾部lastWaiter

public class ConditionObject implements Condition, java.io.Serializable {
    private static final long serialVersionUID = 1173984872572414699L;
    private transient Node firstWaiter;
    private transient Node lastWaiter;
    .....
}

static final class Node {
    // Node 节点就是AQS的同步（CLH列表）的节点
    volatile int waitStatus;
    volatile Node prev;
    volatile Node next;
    volatile Thread thread;
    Node nextWaiter;
    ....
}

这个链表是Condition引入的，也就是我们所说的条件队列，需要配合着ReentranctLock的同步队列一起使用，具体如下图所示：

image

在这里停下了，思考一下！

已经知道了CLH队列是通过双向循环链表去维系各个Node的关系，所有在CLH队列的节点都有可能被唤醒，那么如何实现节点的阻塞暂停呢，节点不放在CLH队列即可，再联想到ConditionObject中说的firstWaiter和lastWaiter节点信息，应该能认识到其大致的工作原理了。

当一个节点通过await阻塞时，需要把这个节点移除出CLH队列，添加到当前的Condition单链表中，但是当其通过signal唤醒时，就需要从Condition单链表中移到CLH队列的尾部，在CLH队列中的节点总有机会被唤醒操作。

await 方法 挂起线程

public final void await() throws InterruptedException {
    if (Thread.interrupted())
        // 感知到线程中断，则直接抛出线程中断移除
        throw new InterruptedException();
    // 添加一个节点到condition条件队列中
    Node node = addConditionWaiter();
    
    // 释放当前的锁，同时记录下state值，以便于后续唤醒操作
    // 如果当前线程重入的2次，则唤醒时也需要重入2次，否则会出现重入不一致的情况使得锁释放和获取出现问题
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    while (!isOnSyncQueue(node)) {
        // 如果当前的节点不在CLH同步队列中，则直接通过park暂停当前线程
        LockSupport.park(this);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            // 出现中断了，跳出break
            break;
    }
    // 到这里来意味着两种情况，1、节点已经移到CLH对着，2、发生中断
    // 被唤醒了，尝试获取锁执行任务操作
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

addConditionWaiter 添加节点

在看这个方法时，需要意识到当前是在一个线程内调用的，是线程安全的

private Node addConditionWaiter() {
    Node t = lastWaiter;
    // If lastWaiter is cancelled, clean out.
    if (t != null && t.waitStatus != Node.CONDITION) {
        // 尾节点存在而取消了，则进行取消节点操作
        unlinkCancelledWaiters();
        t = lastWaiter;
    }
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    // 新增一个节点，是条件队列的类型
    if (t == null)
        firstWaiter = node;
        // 如果尾节点为null，则意味着是初始化的Condition，直接设置头尾节点指向同一个即可
    else
        // 否则采用尾插法，把新建节点插入到当前尾节点后面
        t.nextWaiter = node;
    // 更新尾节点
    lastWaiter = node;
    return node;
}

再来看看如何清除条件队列里面已经被取消的节点信息

private void unlinkCancelledWaiters() {
    Node t = firstWaiter;
    Node trail = null;
    while (t != null) {
        // 从头结点开始扫描
        Node next = t.nextWaiter;
        if (t.waitStatus != Node.CONDITION) {
            // 如果节点状态不是condition，则需要移除
            // 下面的语句则是单链表移除节点的基本操作，就不再说明了
            t.nextWaiter = null;
            if (trail == null)
                firstWaiter = next;
            else
                trail.nextWaiter = next;
            if (next == null)
                lastWaiter = trail;
        }
        else
            trail = t;
        t = next;
    }
}

fullyRelease 释放锁资源

在这里依旧不会出现线程不安全的情况，直到释放锁之后

final int fullyRelease(Node node) {
    boolean failed = true;
    try {
        int savedState = getState();
        // 获取当前锁重入次数
        if (release(savedState)) {
            // 一般情况肯定是能够正常释放锁的，记录下当前次数
            // 而且是会主动的释放并执行当前CLH队列头部节点的
            failed = false;
            return savedState;
        } else {
            // 在已经获取到锁的时候却不能正常的释放锁，提示非法的监视器状态异常
            throw new IllegalMonitorStateException();
        }
    } finally {
        if (failed)
            // 如果释放锁出现错误，则设置节点状态为取消
            // 后续通过unlinkCancelledWaiters方法就会被移除出条件队列的
            node.waitStatus = Node.CANCELLED;
    }
}

现在已经成功的释放了锁资源，加入到了条件队列需要进行挂起操作，不再是线程安全的操作的

final boolean isOnSyncQueue(Node node) {
    if (node.waitStatus == Node.CONDITION || node.prev == null)
        // 如果节点状态是条件节点或者头节点为null 返回false
        // 前节点是在同步队列CLH中才使用的
        return false;
    if (node.next != null) 
        // next节点是同步队列中才使用的，在条件队列中是使用的nextWaiter
        return true;
    // 因为线程不安全，故在这里是采取了从尾部依次遍历判断的
    // 而不是通过node.prev() != null 判断的
    // 因为在进入到同步队列中是若干个节点先进行 node.prev = tail 操作
    // 后面再进行CAS 操作设置为新的tail，但是存在CAS失败的情况
    // 那么就出现了node.prev != null 但是也不在同步队列中的情况，所以不能通过node.prev判断 
    return findNodeFromTail(node);
}

private boolean findNodeFromTail(Node node) {
    Node t = tail;
    // tail 是同步队列的尾部节点信息
    for (;;) {
        if (t == node)
            return true;
        if (t == null)
            // 在同步队列中没有找到该节点信息
            return false;
        // 依次往前判断
        t = t.prev;
    }
}

通过上述的isOnSyncQueue方法再一步确认其节点node是否在同步队列中，当其值返回了false，则进入到while循环中

while (!isOnSyncQueue(node)) {
    LockSupport.park(this);
    // 之前的fullRelease只是释放ReentrantLock的锁，而这一步才是真正的挂起线程
    // 在调用signal方法或者线程发生中断都会进行唤醒操作，执行下面的「在等待中检查中断」方法
    if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
        break;
}

我们先来看看唤醒操作是如何实现的

signal 唤醒线程

public final void signal() {
    // 同样的在调用signal方法前也需要获取到当前的锁
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    if (first != null)
    doSignal(first);
}

private void doSignal(Node first) {
    // first节点是条件队列的首节点
    do {
        if ( (firstWaiter = first.nextWaiter) == null)
            // first节点的下一个节点为null，则表示到条件队列的尾部了，直接设置尾节点为null
            lastWaiter = null;
        // 剥离当前first节点和条件队列的关系
        first.nextWaiter = null;
    } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
    // 如果转移转移当前节点失败，则转移下一个节点
}
        
final boolean transferForSignal(Node node) {
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            // CAS 设置当前节点的状态失败，说明该节点已经被取消了，转移下一个节点
            // 还记得fullyRelease中设置节点状态为Node.CANCELLED的么？
            return false;
        // 否则设置当前node节点的状态为0
            
        Node p = enq(node);
        // 采取CAS + while的方式进入到同步队列中，且p节点是node节点的前置节点
        // 虽然这里采取了加锁操作进入到同步队列，但是也存在其他节点插入到同步队列的情况
        int ws = p.waitStatus;
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
            // ws > 0 表示 node前置节点p设置了取消状态，应该唤醒node节点线程
            // 否则w是<= 0 的情况下，需要设置前置节点状态为-1
            
            // 如果前置节点取消或者 CAS设置前置节点状态失败，则唤醒当前的node线程
            LockSupport.unpark(node.thread);
        return true;
}

那么唤醒是在哪里呢？答案在isOnSyncQueue的while循环中

唤醒操作

while (!isOnSyncQueue(node)) {
    LockSupport.park(this);
    // 在这里被唤醒，继续执行操作
    if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
        break;
}

private int checkInterruptWhileWaiting(Node node) {
    // Thread.interrupted() 获取当前线程状态，并重置中断状态
    return Thread.interrupted() ?  (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :0;
    // 线程是否发生中断？线程中断是发生在await之前还是之后？
    // 线程未发生中断 返回0
    // 线程在await期间发生中断 返回 REINTERRUPT 1
    // 线程在await之后 返回 THROW_IE -1  
}
        
final boolean transferAfterCancelledWait(Node node) {
    if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
        // CAS 成功的讲当前node节点修改为状态0，然后进入到同步队列中，等待被调用执行
        // 那么也就意味着是在signal方法前发生的，signal方法会一开始就进行状态修改为0的CAS操作
        // 即使发生中断，依旧会把当前节点转移到同步队列中的
        enq(node);
        return true;
    }
    
    // CAS 设置状态失败了，肯定是因为signal方法已经设置为0了，需要严格确保node节点进入到同步队列中
    while (!isOnSyncQueue(node))
        Thread.yield();
    // 返回了false，是在signal之后，转移到同步队列完成之前出现了中断操作
    return false;
}

唤醒后获取锁

while循环跳出来有两种情况，1、被signal正常唤醒，2、因中断被唤醒，其中interruptMode记录着因为哪种情况导致的中断

• 0：表示无中断发生
• REINTERRUPT：signal发生之后，转移到同步队列中之前
• THROW_IE：signal发生之前

if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;

现在开始尝试着获取锁，savedState表示着这个线程之前的重入次数，现在需要重新恢复到之前的重入次数，当acquireQueued方法返回true意味着出现中断了，同时 interruptMode != THROW_IE说明了，则需要重置中断状态

if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();

问题来了，此时不是已经唤醒操作了么，节点肯定是在同步队列的，node.nextWaiter 是为null的，答案是不一定

在doSignal方法中存在first.nextWaiter = null;的操作，但是中断发生在signal方法调用前就会出现first.nextWaiter!= null的情况，此时节点虽然已经移动到同步队列中，但是nextWaiter还和条件队列挂在一起，通过unlinkCancelledWaiters彻底打断同条件队列的联系

中断处理

if (interruptMode != 0)
    reportInterruptAfterWait(interruptMode);
    
private void reportInterruptAfterWait(int interruptMode) throws InterruptedException {
    if (interruptMode == THROW_IE)
        // 中断是signal方法前出现的，也就是await期间出现异常，直接抛出异常
        throw new InterruptedException();
    else if (interruptMode == REINTERRUPT)
        // signal 之后出现的中断请求，重新触发该线程的中断事件
        // Thread.currentThread().interrupt();
        selfInterrupt();
 }

到现在基本上整个条件队列的大致源码已经分析完成了，当然不得不提的是条件队列提供的方法不止await和signal，共包含如下方法

• public final void await() throws InterruptedException
• public final boolean await(long time, TimeUnit unit) throws InterruptedException 返回是否超时
• public final long awaitNanos(long nanosTimeout) throws InterruptedException 返回超时的具体情况
• public final void awaitUninterruptibly()
• public final boolean awaitUntil(Date deadline) throws InterruptedException
• public final void signalAll()
• public final void signal()

添加了超时的等待设置，则是在释放锁之后，在while循环中去判断是否发生超时情况，如果发生超时情况则break（相当于除了正常signal调用、中断之外还有超时这个因素导致的唤醒操作），此外pack也改成了parkNanos方法了

awaitUninterruptibly 则不考虑具体的异常是由谁造成的，也不会抛出InterruptedException异常，而是触发中断事件，由线程自身决定

signalAll则和signal方法不同，signal会沿着条件队列查找出第一个可以唤醒的节点去唤醒，而signalAll则是通过批量唤醒所有的节点，而不考虑节点是否能够唤醒，具体代码如下，也很简单的

private void doSignal(Node first) {
    do {
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        first.nextWaiter = null;
    } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
}

private void doSignalAll(Node first) {
    lastWaiter = firstWaiter = null;
    do {
        Node next = first.nextWaiter;
        first.nextWaiter = null;
        transferForSignal(first);
        first = next;
    } while (first != null);
}