并发编程学习笔记（十五、AQS同步器源码解析4，AQS条件锁Condition实现原理）

目录：

• 概述
• 同步队列、条件队列
• Condition源码解析

概述

首先我们知道Condition的await()、signa()是内置锁synchronize配套的wait()及notify()的增强，可以更加细化的控制锁的唤醒条件。

那么我们这里来类比下它们之间的机制：

• 同步：内置锁的wait()必须要在synchronize代码块中才能调用，也就是必须要获取到监视器锁之后才能执行；await()与之类似，也是必须获取到锁资源后才能执行。
• 等待：调用wait()的线程会释放已经获取的锁，进入当前监视器锁的等待队列中；与之类似，await()也会释放自己的lock，进入到当前Condition的条件队列中。
• 唤醒：调用notify()会唤醒等待在该监视器锁的线程，并参与监视器锁的竞争，并在获取锁之后从wait()处恢复执行；同理，调用signal()会唤醒对应线程，并获取锁资源，在获取后会在signal()处恢复执行。

同步队列、条件队列

上面我们说到了内置锁与Condition的等待，一个是放到等待队列中，一个是放到条件队列中，那么我们这里来分别说下。

——————————————————————————————————————————————————————————————————————

1、同步队列（等待队列）：

前面我们在说独占锁的获取时，所有等待的线程都会被包装成一个Node进入同步队列。

由图可知其是一个先进先出（FIFO）的同步阻塞队列，基于双向链表实现，使用prev、next来串联前驱及后继节点。

其中有一个nextWaiter属性一直没有提到过，即使是共享模式下它也只是简单的标记了一下，指向了一个空节点。

因此在同步队列中，并不会用它来串联节点。

——————————————————————————————————————————————————————————————————————

2、条件队列：

Condition同理，也会在调用await()时将其包装成一个Node放入到条件队列中，但值得注意的是这个条件队列是基于单向链表实现，以nextWaiter来标记下一个节点的指针。

与同步队列恰好相反，条件队列并不会使用prev、next来串联链表，而是使用nextWaiter。

其结构大致如下：

你还记得AQS中Node的waitStatus的值嘛，没错有一个CONDITION就是来标记同步队列的，在源码中我会详述。

——————————————————————————————————————————————————————————————————————

3、同步队列与条件队列的联系：

一般情况下同步队列和条件队列是互相独立，没有任何关系的。但当调用某个条件队列中的signal()去唤醒线程时，会让这个线程和普通线程一样去竞争锁，如果没有抢到就会被加到同步队列中去，此时这个条件队列就会变成同步队列了。

这里你可能会有一个疑问，为啥条件队列获取不到锁后不是重新回到条件队列，而是加入到同步队列去了呢，这不是有问题嘛。

其实是没有问题的，因为你是满足条件后才会调用signal()唤醒线程的，所以线程尝试获取锁的时候其实已经是满足唤醒条件了，就算未获取到锁也只是重新入队（同步队列）而已，照样可以在下次被唤醒时重新竞争锁。

这里我们需要注意的是Node节点是一个一个转移过去的，即使是调用signalAll()也是一样，而不是将整个条件队列接在同步队列的队尾。

那为什么是一个一个的转移呢，因为同步队列使用的是prev、next来串联链表，条件队列使用的是nextWaiter，它们在数据结构上是完全独立的。因此我们在转换的过程中需要把nextWaiter断开，建立新的prev、next，这也就是为啥要一个个转移的原因之一了。

——————————————————————————————————————————————————————————————————————

4、入队和出队的锁状态：

• 条件队列：入队前有锁 >>> 队列中释放锁 >>> 离开队列竞争锁 >>> 转到同步队列。
• 同步队列：入队前无锁 >>> 队列中竞争锁 >>> 离开队列持有锁。

好了，通过上面的介绍相信你对条件对口已经有个一个感性的认知，接下来我们来说说它的源码。

Condition源码解析

1、基本属性及构造函数：

前面我们说到Condition接口的实现类就是ConditionObject，在学习其主要逻辑前我们还是照常先看下其基本属性及构造函数吧。

/** 条件队列队头节点 */
private transient Node firstWaiter;
/** 条件队列队尾节点 */
private transient Node lastWaiter;

/**
 * 空的构造器，没啥好说的，但可以看出条件队列是延时初始化的。
 */
public ConditionObject() { }

2、await()：

然后我们来看下Condition的线程阻塞方法await()。

public final void await() throws InterruptedException {
    // 若当前线程已经中断则抛出异常
    if (Thread.interrupted())
        throw new InterruptedException();
    // 将当前线程封装成Node添加到条件队列中
    Node node = addConditionWaiter();
    // 释放当前线程所占用的锁，并保存当前锁的状态
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    // 若当前线程不在同步队列中，说明刚刚被await了（也就是条件队列），但还未调用signal()方法，则直接循环阻塞线程
    while (!isOnSyncQueue(node)) {
        // 阻塞线程，停止运行
        LockSupport.park(this);
        // 执行到这说明，要么已经调用signal()方法，要么是线程中断了
        // 所以这里检查下唤醒原因，如果是因中断而唤醒，则跳出循环
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    // 这部分非主流程，之后再解析
    /*if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);*/
}

上面已经把await()的主要流程说明了下，其核心步骤如下，接下来我会一一解析：

• 将当前线程封装成Node添加到条件队列中
• 释放当前线程所占用的锁，并保存当前锁的状态
• 循环阻塞线程

——————————————————————————————————————————————————————————————————————

A、将当前线程封装成Node添加到条件队列中：addConditionWaiter()

private Node addConditionWaiter() {
    // 获取队尾队列
    Node t = lastWaiter;
    // 如果队尾队列不是CONDITION状态，则遍历整个队列，清除所有被cancel的节点
    if (t != null && t.waitStatus != Node.CONDITION) {
        // 清除所有已经取消等待的线程
        unlinkCancelledWaiters();
        t = lastWaiter;
    }
    // 将当前队列包装成CONDITION的Node，并传入条件队列
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    if (t == null)
        // 若队尾为空，说明队列还无节点，此时将当前节点设置为头节点
        firstWaiter = node;
    else
        // 队尾非空，将队尾节点的nextWaiter指针指向当前节点，也就是当当前节点入到队尾
        t.nextWaiter = node;
    lastWaiter = node;
    return node;
}

这一步非常简单，而unlinkCancelledWaiters()也不过是对链表的操作，它剔除了队列中状态不为CONDITION的节点，只要了解链表的话可以非常轻松的读懂，我就不再赘述。

private void unlinkCancelledWaiters() {
    Node t = firstWaiter;
    Node trail = null;
    while (t != null) {
        Node next = t.nextWaiter;
        if (t.waitStatus != Node.CONDITION) {
            t.nextWaiter = null;
            if (trail == null)
                firstWaiter = next;
            else
                trail.nextWaiter = next;
            if (next == null)
                lastWaiter = trail;
        }
        else
            trail = t;
        t = next;
    }
}

这里有一个值得注意的问题，就是是否存在两个不同的线程同时入队的情况。

答案是否定的，因为入队前是一定获取到锁的，所以不存在并发的情况，不需要CAS操作。

——————————————————————————————————————————————————————————————————————

B、释放当前线程所占用的锁，并保存当前锁的状态：fullyRelease(node)

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;
    }
}

首先我们要知道，当调用fullyRelease()方法时说明线程已经入队了。然后根据第5行代码可以看出，其是根据release()方法来释放锁的，这个之前介绍过就不再赘述。

但有一点奇怪的是，明明只是释放当前线程占用的锁，那为啥函数名却叫fullyRelease呢，这是因为对于可重入锁来说，无论重入了所少次，都是一次性释放完的。

这个地方有一点需要注意下，就是调用release方法是可能会抛出IllegalMonitorStateException异常，这是因为当前线程可能并不是持有锁的线程。

可之前不是说调用await方法的线程一定是持有锁的嘛，哈哈话虽这样说，但其实await方法是并没有校验的，而是放到了fullyRelease()的tryRelease()来校验Thread.currentThread() == getExclusiveOwnerThread()。

如ReentrantLock：

protected final boolean tryRelease(int releases) {
    int c = getState() - releases;
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}

——————————————————————————————————————————————————————————————————————

C、循环阻塞线程：

在线程锁释放完毕后，就会调用LockSupport.park(this)方法将当前线程挂起，等待被signal()方法唤醒。

但在挂起线程前为啥要需要调用isOnSyncQueue()来确保这个不在同步队列中呢，之前不是说了同步队列和条件队列在数据结构上述无关的嘛。为啥会出现在同步队列的情况，这是因为可能其它线程获取到了锁，导致并发调用了signal（下文说明signal会详述）。

final boolean isOnSyncQueue(Node node) {
    // 当前线程状态为CONDITION或前驱节点为空（条件队列的prev、next都是空）则不为同步队列
    if (node.waitStatus == Node.CONDITION || node.prev == null)
        return false;
    // 若node的后继节点非空，则肯定为同步队列的节点
    if (node.next != null) // If has successor, it must be on queue
        return true;
    // 从尾部向前搜索
    return findNodeFromTail(node);
}

private boolean findNodeFromTail(Node node) {
    // 获取尾部节点tail，只有在同步队列中tail才非空，条件队列的队尾是lastWaiter节点
    Node t = tail;
    for (;;) {
        if (t == node)
            return true;
        if (t == null)
            return false;
        // 从尾部向前遍历队列
        t = t.prev;
    }
}

——————————————————————————————————————————————————————————————————————

3、signal()：

唤醒方法分为signal()，唤醒单个线程；和signalAll()，唤醒多个线程。因signal和signalAll非常相似，所以我先说说signalAll()。

signalAll()：

首先在看signalAll源码之前，我们先要知道调用signalAll方法的线程和signAll方法要唤醒的线程（在条件队列里等待的线程）的区别：

• 调用signalAll方法的线程本身已经持有锁了，现在准备去释放锁。
• signalAll方法要唤醒的线程是是条件队列里挂起了，等待被signal唤醒，然后去竞争锁。

——————————————————————————————————————————————————————————————————————

然后我们来看看signalAll的源码：

public final void signalAll() {
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    if (first != null)
        doSignalAll(first);
}

1、首先走了isHeldExclusively()逻辑，如果不满足则抛出监视器锁状态异常，我们通过查阅其源码可发现isHeldExclusively()内部的实现是直接抛异常的，也就是说这个方法肯定是交给子类实现的。

protected boolean isHeldExclusively() {
    throw new UnsupportedOperationException();
}

那我们来看看其子类ReentrantLock，可以发现它回去拿当前线程与当前持有锁的线程做比较。

为什么会这样比较呢，我们之前不是说到了嘛，Condition其实就是Object中wait和notify方法的扩展。

所以调用signalAll的肯定也是需要获取到锁的，哈哈。

protected final boolean isHeldExclusively() {
    return getExclusiveOwnerThread() == Thread.currentThread();
}

——————————————————————————————————————————————————————————————————————

2、然后我们将队头的节点交给first，并在其非空的情况下（也就是条件队列有数据存在）调用doSignalAll()。

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

首先将队首队尾的指针都指向空，也就是清空条件队列（因为是signalAll嘛，唤醒所有，当然是情况队列咯）。

紧接着遍历整个条件队列，因为first之前的指针是指向firstWaiter的，所以是队头节点，它能够遍历整个队列。

最后会依次调用transferForSignal，一个个添加到同步队列的队尾。

final boolean transferForSignal(Node node) {
    // 先通过CAS将node节点的状态由CONDITION置为0，若修改失败说明该节点已经被CONDITION了，则直接返回操作下一个节点
    // 如果操作成功，说明已经将该节点从条件队列唤醒了，因此会被添加到条件队列的尾队等待竞争锁
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;

    // 自旋的添加到同步队列队尾，并拿到当前节点的前驱节点
    Node p = enq(node);
    int ws = p.waitStatus;
    // 将当前节点的前驱节点的状态置为SIGNAL，也就是要前驱节点来唤醒当前节点
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);
    return true;
}

——————————————————————————————————————————————————————————————————————

在说signal方法前我们先来总结下signalAll：

• 将条件队列清空（只是令lastWaiter = frstWaliter = null，队列中的结点和连接关系仍然还存在）。
• 将条件队列中的头结点取出，使之成为孤立结点（nextWaiter的prev和next属性都为null）。
• 如果该结点处于被Cancelled了的状态，则直接跳过该结点（由于是孤立结点，则会被GC回收）。
• 如果该结点处于正常状态，则通过enq()法将 它添加到同步队列的末尾。
• 判断是否需要将该结点唤醒（包括设置该结点的前驱结点的状态为SIGNAL），如有必要，直接唤醒该结点。
• 重复2-5，直到整个条件队列中的结点都被处理完。

——————————————————————————————————————————————————————————————————————

signal()：

弄懂了signalAll后signal应该会很轻松，因为它们大同小异，signal只是唤醒队列中第一个没有被Cancelled的线程，如果没有则将条件队列清空。

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);
}