谈谈JUC----------Condition源码分析

1、什么是 Condition？

之前多线程通讯使用的是Object类提供的wait以及notify方法，Condition也有相似wait、notify的功能，且提供了当线程中断时是否作出相应的处理方法。Condition底层实现基于FIFO队列，它必须结合Lock锁一块儿使用，而且Condition实例由Lock建立。最后总结起来就是，Condition是一种多线程通讯工具，表示多线程下参与数据竞争的线程的一种状态，主要负责多线程环境下对线程的挂起和唤醒工做。

下面贴出一张截取自《Java并发编程的艺术》的对比图：

2、Condition 源码分析

一、Condition架构及其内部结构是怎样的？

Condition是一个接口，jdk中提供了Condition的一个实现类ConditionObject,ConditionObject是AQS中的内部类，由于Condition的操做须要获取相关的锁，而AQS又是实现同步锁的基础。Condition提供了下面几种操做方法：

• void await() throws InterruptedException

当调用这个方法时，线程将被挂起，直到被其余线程唤醒为止。注意当挂起的线程被中断时，将抛出InterruptedException异常。

• void awaitUninterruptibly()

此方法与上面的await()方法相似，区别是该方法不处理线程中断的状况。

• long awaitNanos(long nanosTimeout) throws InterruptedException

表示能够最长等待指定时间，除非中途被中断或者提早唤醒了，返回值=nanosTimeout-已等待的时间。

• void awaitUninterruptibly()

此方法与上面的await()方法相似，区别是该方法不处理线程中断的状况。

• boolean await(long time, TimeUnit unit) throws InterruptedException

同awaitNanos(),只不过能够指定时间单位。

• boolean awaitUntil(Date deadline) throws InterruptedException

表示线程挂起，知道某个时间点唤醒该线程。

• void signal()以及void signalAll()

唤醒操做。

那么问题来了，怎么获取Condition对象呢？查看Lock接口，该接口声明了一个newCondition()方法，返回一个Condition对象，如Lock接口的一个实现ReentrantLock中则提供了该方法的实现，但本质上仍是建立了ConditionObject对象。

源码以下：

public Condition newCondition() {
    return sync.newCondition();
}
复制代码

其中sync为继承自AQS的同步器。

final ConditionObject newCondition() {
    return new ConditionObject();
}
复制代码

关于AQS，能够参考下面这篇文章：

Java 并发编程 ----- AQS（抽象队列同步器）

二、Condition中await内部实现细节

下边为await()方法源代码：

public final void await() throws InterruptedException {
    // 一、线程若是中断，那么抛出异常
    if (Thread.interrupted())
        throw new InterruptedException();
    // 二、将当前线程包装成为一个Node节点，加入FIFO队列中
    Node node = addConditionWaiter();
    // 三、释放锁
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    // 四、判断节点是否在同步队列（注意非Condition队列）中，若是没有，则挂起当前线程，由于该线程还没有具有数据竞争资格
    while (!isOnSyncQueue(node)) {
        // 五、挂起线程
        LockSupport.park(this);
        // 六、中断直接返回
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    // 七、参与数据竞争（非中断时执行）
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
        
    // 清理条件队列中状态为cancelled的节点
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
    }
复制代码

从上面代码咱们能够看出，由于await自己不支持中断，所以若是当前线程被中断了，那么第一步直接抛出异常。第二步将当前线程包装成一个Node节点，加入到Condition条件队列中，Node与AQS使用的是同一个类型，查看AQS可知，每一个Node与一个当前Thread相关联。

static final class Node {
    volatile Thread thread;
}
复制代码

接着执行第三步释放锁，这里就涉及到为何说Condition使用时必须先得到锁的问题了，下方为fullyRelease()源码：

final int fullyRelease(Node node) {
    boolean failed = true;
    try {
        int savedState = getState();
        // 这里直接执行释放锁操做
        if (release(savedState)) {
            failed = false;
            return savedState;
        } else {
            // 释放失败，则报错
            throw new IllegalMonitorStateException();
        }
    } finally {
        if (failed)
            node.waitStatus = Node.CANCELLED;
    }
}
复制代码

第4-6步则不断循环，直到该节点出如今同步队列中，注意同步队列不是条件队列，一个属于AQS，一个属于Condition,那么，这个while循环是怎样判断是否在同步队列中的呢？其实很简单，就是当把节点的状态改为非CONDITION就能够了，好比调用了signal(),并且请注意，LockSupport.park(this);这句代码执行完以后是阻塞代码了哈，后面分析signal会提到。源码以下：

final boolean isOnSyncQueue(Node node) {
    // 判断节点的状态是否是CONDITION，CONDITION表示该节点正在处于等待某个条件，此时就应该park挂起线程
    if (node.waitStatus == Node.CONDITION || node.prev == null)
        return false;
    if (node.next != null) // If has successor, it must be on queue
        return true;
    return findNodeFromTail(node);
}
复制代码

第7步也很好理解，既然当前线程等待的条件都有了，被唤醒了，那么就直接参与数据竞争就完事了，第8步清理掉一些状态为cancelled的节点，线程因为中断或超时时，节点的状态就会被标记成cancelled

三、Condition中signal内部实现细节

public final void signal() {
    // 一、必须得到锁
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    if (first != null)
        doSignal(first);
}
复制代码

private void doSignal(Node first) {
    do {
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        first.nextWaiter = null;
        // 将节点插入到同步队列中
    } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
}
复制代码

调用condition的signal的前提条件是当前线程已经获取了lock，该方法会使得等待队列中的头节点即等待时间最长的那个节点移入到同步队列，而移入到同步队列后才有机会使得等待线程被唤醒，即从await方法中的LockSupport.park(this)方法中返回（由于上文提到这个方法是阻塞住当前代码的），从而才有机会使得调用await方法的线程成功退出

3、总结

使用锁结合Condition能够很好的解决生产者消费者问题，须要注意的就是，使用Condition时必须先获取锁，不然将报错，也就是说，通常会按下面方式去调用Condition。

lock.lock();
condition.await();

condition.signal();
lock.unlock();
复制代码