本文共 18633 字，大约阅读时间需要 62 分钟。

上篇以 ReentrantLock 的非公平锁入手，分析了 AQS 独占锁的获取和释放，接下来以公平锁分析 AQS 中的 Condition。

公平锁与非公平锁

先说下什么是公平锁：获取锁之前，先判断阻塞队列中是否有节点，有则加入到队尾，没有节点就开始竞争锁。像食堂排队一样，大家按照次序排成一队，后面来的人排在队尾，不许插队。在绝对时间上，先申请的先获取到锁，这就是公平。

公平锁源代码：

/** * Sync object for fair locks */static final class FairSync extends Sync {
       private static final long serialVersionUID = -3000897897090466540L;    final void lock() {
       	//区别1        acquire(1);    }    /**     * Fair version of tryAcquire.  Don't grant access unless     * recursive call or no waiters or is first.     */    protected final boolean tryAcquire(int acquires) {
           final Thread current = Thread.currentThread();        int c = getState();        if (c == 0) {
           	//区别2，没有前驱节点才获取锁            if (!hasQueuedPredecessors() &&                compareAndSetState(0, acquires)) {
                   setExclusiveOwnerThread(current);                return true;            }        }        else if (current == getExclusiveOwnerThread()) {
               int nextc = c + acquires;            if (nextc < 0)                throw new Error("Maximum lock count exceeded");            setState(nextc);            return true;        }        return false;    }        //队列有前驱返回true，没有则返回false代表可以竞争锁    public final boolean hasQueuedPredecessors() {
           // The correctness of this depends on head being initialized        // before tail and on head.next being accurate if the current        // thread is first in queue.        Node t = tail; // Read fields in reverse initialization order        Node h = head;        Node s;        //头节点一定不是自己，关键在于判断头节点的后驱是不是自己        //头尾相等，表示队列中只有一个节点，这个节点可能马上就要执行完了，直返返回false表示没有前驱        //头尾不相等，头节点的后驱为空（第一次初始化队列，但是其他线程先自己一步设置了头），返回true        //头尾不相等，头节点的后驱不是自己，返回true        return h != t &&            ((s = h.next) == null || s.thread != Thread.currentThread());    }}

非公平锁源代码：

/** * Sync object for non-fair locks */static final class NonfairSync extends Sync {
       private static final long serialVersionUID = 7316153563782823691L;    /**     * Performs lock.  Try immediate barge, backing up to normal     * acquire on failure.     */    final void lock() {
       	//区别1，直接先竞争一次锁        if (compareAndSetState(0, 1))            setExclusiveOwnerThread(Thread.currentThread());        else            acquire(1);    }    protected final boolean tryAcquire(int acquires) {
           return nonfairTryAcquire(acquires);    }        /**     * Performs non-fair tryLock.  tryAcquire is implemented in     * subclasses, but both need nonfair try for trylock method.     */    final boolean nonfairTryAcquire(int acquires) {
           final Thread current = Thread.currentThread();        int c = getState();        if (c == 0) {
           	//区别2，不讲道理，直接获取锁            if (compareAndSetState(0, acquires)) {
                   setExclusiveOwnerThread(current);                return true;            }        }        else if (current == getExclusiveOwnerThread()) {
               int nextc = c + acquires;            if (nextc < 0) // overflow                throw new Error("Maximum lock count exceeded");            setState(nextc);            return true;        }        return false;    }}

看代码也很容易理解，公平锁和非公平锁的差别在两个地方：

• 非公平锁 lock() 的时候，先尝试获取锁，失败再调用 acquire(1) ，而公平锁没有这一步。
• 公平锁的 tryAcquire() 方法中调用了 !hasQueuedPredecessors() 判断队列是否有节点，而非公平锁每次获取锁的时候没有判断前驱接节点，没有即代表不公平，插队了。

Condition 介绍

如果说 ReentrantLock 是 synchronized 的替代选择，Condition 则是将 wait、notify、notifyAll 等操作转化为相应的对象行为，直接可控。

看下 Doug Lea 提供的例子：假设我们有一个有界缓冲区支持 put 和 take 方法，如果一个 take 试图在一个空缓冲区中取数据，则线程将被阻塞，直到新加入一个数据；如果 put 试图向已经满了的缓冲区中存数据，那么该线程将被阻塞，直到有空间可用。这可通过两个 Condition 实例实现。

/**  * @since 1.5 * @author Doug Lea */public class BoundedBuffer {
       final Lock lock = new ReentrantLock();    final Condition notFull = lock.newCondition();    final Condition notEmpty = lock.newCondition();    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();            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();            Object x = items[takeptr];            if (++takeptr == items.length) takeptr = 0;            --count;            notFull.signal();            return x;        } finally {
               lock.unlock();        }    }}

同步器可以创建多个 Condition，每一个 Condition 包含一个等待队列，Condition 含有首尾两个节点的引用。等待队列复用同步队列中的的节点，它们都是 AQS 的静态内部类 Node。

同步队列中的首节点获取锁后，如果调用 Condition 的 await() 方法，那么当前获取锁的节点会从同步队列移动到等待队列的队尾；同步队列中获取锁的节点调用 Condition 的 signal() 方法，那么等待队列中的头结点会从等待队列移动到同步队列的末尾。

ConditionObject 类定义，reentrantLock.newCondition() 是如何新建一个 Condition 的，它实际最终创建的是 AQS 的内部类 ConditionObject ，ConditionObject 实现了接口 Condition。

public Condition newCondition() {
       return sync.newCondition();}final ConditionObject newCondition() {
       return new ConditionObject();}public class ConditionObject implements Condition, java.io.Serializable {
       private static final long serialVersionUID = 1173984872572414699L;    /** First node of condition queue. */    //等待队列的第一个节点    private transient Node firstWaiter;    /** Last node of condition queue. */    //等待队列的尾节点    private transient Node lastWaiter;    /**     * Creates a new {@code ConditionObject} instance.     */    public ConditionObject() {
    }}

接下来分析下 condition.await() 是怎么释放的锁，以及怎么将节点从同步队列移动到等待队列中去的。

/** * Implements interruptible condition wait. * 
   
     * 
    
 If current thread is interrupted, throw InterruptedException. * 
    
 Save lock state returned by {@link #getState}. * 
    
 Invoke {@link #release} with saved state as argument, *      throwing IllegalMonitorStateException if it fails. * 
    
 Block until signalled or interrupted. * 
    
 Reacquire by invoking specialized version of *      {@link #acquire} with saved state as argument. * 
    
 If interrupted while blocked in step 4, throw InterruptedException. * 
   
 */public final void await() throws InterruptedException {
   	//进入方法前先判断下线程是否被中断，注意await方法申明是响应中断的    if (Thread.interrupted())        throw new InterruptedException();    //新建一个节点，并加入到等待队列队尾    Node node = addConditionWaiter();    //开始释放锁，并返回之前获取的锁状态，便于之后继续获取等量的重入锁    int savedState = fullyRelease(node);    int interruptMode = 0;    while (!isOnSyncQueue(node)) {
           LockSupport.park(this);        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);}/** * Adds a new waiter to wait queue. * @return its new wait node */private Node addConditionWaiter() {
   	//线程是拥有锁的，不需要考虑并发    Node t = lastWaiter;    // If lastWaiter is cancelled, clean out.    //判断等待队列尾节点是否被取消    if (t != null && t.waitStatus != Node.CONDITION) {
       	//被取消循环的话，循环清除        unlinkCancelledWaiters();        t = lastWaiter;    }    //新建一个节点，节点状态为 Node.CONDITION，代表节点为等待队列    //其他线程调用 signal() 会将这个节点移动到同步队列    Node node = new Node(Thread.currentThread(), Node.CONDITION);    if (t == null)        firstWaiter = node;    else        t.nextWaiter = node;    //设置为队尾    lastWaiter = node;    return node;}Node(Thread thread, int waitStatus) {
    // Used by Condition    this.waitStatus = waitStatus;    this.thread = thread;}/** * Unlinks cancelled waiter nodes from condition queue. * Called only while holding lock. This is called when * cancellation occurred during condition wait, and upon * insertion of a new waiter when lastWaiter is seen to have * been cancelled. This method is needed to avoid garbage * retention in the absence of signals. So even though it may * require a full traversal, it comes into play only when * timeouts or cancellations occur in the absence of * signals. It traverses all nodes rather than stopping at a * particular target to unlink all pointers to garbage nodes * without requiring many re-traversals during cancellation * storms. */private void unlinkCancelledWaiters() {
   	//从头节点开始遍历整个队列，更新头尾节点的指针    Node t = firstWaiter;    //记录上次正常的节点    Node trail = null;    while (t != null) {
           Node next = t.nextWaiter;        //被取消就不是进入队列时的状态 Node.CONDITION 了，这时需要从队列中移除        if (t.waitStatus != Node.CONDITION) {
               t.nextWaiter = null;            if (trail == null)                firstWaiter = next;            else                trail.nextWaiter = next;            if (next == null)                lastWaiter = trail;        }        else        	//节点正常，记录本节点到trail            trail = t;        t = next;    }}/** * Invokes release with current state value; returns saved state. * Cancels node and throws exception on failure. * @param node the condition node for this wait * @return previous sync state */final int fullyRelease(Node node) {
       boolean failed = true;    try {
           int savedState = getState();        //这个方法参考上篇文章分析，释放锁并唤醒下个节点        if (release(savedState)) {
               failed = false;            return savedState;        } else {
           	//释放锁失败，没有完全释放完（ state 不为0，可能 tryRelease() 方法实现的有问题），抛异常            throw new IllegalMonitorStateException();        }    } finally {
       	//正常情况下，failed 标识为 false        if (failed)    		//可能没有持有锁或没有完全释放完        	//既然么有释放锁成功，将此节点标记为取消，待后续新加入的节点移除            node.waitStatus = Node.CANCELLED;    }}

回到 await() 方法，上边的步骤只是加入到了等待队列并释放了锁，下面开始判断自己是否在同步队列，不在需要被阻塞，然后等待别人将自己唤醒并加入到同步队列。

public final void await() throws InterruptedException {
       if (Thread.interrupted())        throw new InterruptedException();    Node node = addConditionWaiter();    int savedState = fullyRelease(node);    int interruptMode = 0;    //判断节点是否在同步队列，加入同步队列需要靠其他线程调用 condition.signal()    while (!isOnSyncQueue(node)) {
       	//不在同步队列，阻塞        LockSupport.park(this);        //恢复执行之后，首先检查等待期间是否有被中断        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)            break;    }    //被中断或者进入阻塞队列之后被唤醒 都会跳出上面的循环    //尝试获取锁，acquireQueued() 方法也可被中断，且会返回一个中断标识    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)        interruptMode = REINTERRUPT;    //获取锁之后，检查下是否还有后续等待节点，清除被取消的    if (node.nextWaiter != null) // clean up if cancelled        unlinkCancelledWaiters();    if (interruptMode != 0)    	//处理中断标识        reportInterruptAfterWait(interruptMode);}/** * Returns true if a node, always one that was initially placed on * a condition queue, is now waiting to reacquire on sync queue. * @param node the node * @return true if is reacquiring */final boolean isOnSyncQueue(Node node) {
   	//节点状态还是 Node.CONDITION 表明还在条件队列，没有被 transfer	//每一个进入阻塞队列的节点前驱一定存在，看 enq() 方法，CAS 设置尾成功一定可以看到 node.prev = t	//node.prev = t;	//if (compareAndSetTail(t, node)) {
       if (node.waitStatus == Node.CONDITION || node.prev == null)        return false;    //next 引用只有阻塞队列中的节点才会用到    if (node.next != null) // If has successor, it must be on queue        return true;    /*     * node.prev can be non-null, but not yet on queue because     * the CAS to place it on queue can fail. So we have to     * traverse from tail to make sure it actually made it.  It     * will always be near the tail in calls to this method, and     * unless the CAS failed (which is unlikely), it will be     * there, so we hardly ever traverse much.     */    //可能进入阻塞队列队尾竞争激烈，此时还没能进入阻塞队列    //（只执行了 enq 中的 node.prev = t），再遍历一下队列确定下    //虽然最终一定会进入阻塞队列，但这里不能直接返回true    //因为：假如此节点进入队列前有另外的节点和它竞争先进入队列中，此节点后进入    //如果直接返回，此线程获取到了锁，会设置当前节点为阻塞队列头节点，那么先它之间进入的节点都会被抛弃    return findNodeFromTail(node);}/** * Returns true if node is on sync queue by searching backwards from tail. * Called only when needed by isOnSyncQueue. * @return true if present */private boolean findNodeFromTail(Node node) {
   	//从尾节点开始找 node，找到了就返回 true，不存在则返回 false    Node t = tail;    for (;;) {
           if (t == node)            return true;        if (t == null)            return false;        t = t.prev;    }}/** * Checks for interrupt, returning THROW_IE if interrupted * before signalled, REINTERRUPT if after signalled, or * 0 if not interrupted. */private int checkInterruptWhileWaiting(Node node) {
       return Thread.interrupted() ?    	//线程被中断有两种情况，一种是发生在被唤醒前（进入阻塞队列前），返回 THROW_IE     	//另一种是发生在进入阻塞队列后，返回 REINTERRUPT    	//怎么判断在下面这个方法        (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :        0;}/** * Transfers node, if necessary, to sync queue after a cancelled wait. * Returns true if thread was cancelled before being signalled. * * @param node the node * @return true if cancelled before the node was signalled */final boolean transferAfterCancelledWait(Node node) {
   	//什么时候 status 为 Node.CONDITION？中断发生在下面 park 的地方    /** ....    while (!isOnSyncQueue(node)) {       	LockSupport.park(this);    ...*/    if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
       	//CAS能够执行成功，表明中断发生在其他线程执行 condition.signal() 前    	//节点虽在进入阻塞队列前就被中断，但是必须还是要入队列        enq(node);        return true;    }    /*     * If we lost out to a signal(), then we can't proceed     * until it finishes its enq().  Cancelling during an     * incomplete transfer is both rare and transient, so just     * spin.     */    //中断发生在执行 transferForSignal() 方法中的 compareAndSetWaitStatus(node, Node.CONDITION, 0) 之后    //状态不为 Node.CONDITION 不一定就一定在队列，可能正在进行中    //自循环，等待入队列，transferForSignal() 一定会将 node 入队列    while (!isOnSyncQueue(node))        Thread.yield();    return false;}/** * Throws InterruptedException, reinterrupts current thread, or * does nothing, depending on mode. */private void reportInterruptAfterWait(int interruptMode)    throws InterruptedException {
       if (interruptMode == THROW_IE)        throw new InterruptedException();    else if (interruptMode == REINTERRUPT)    	//重设中断标识        selfInterrupt();}

signal() 方法是怎么将调用 await() 方法被阻塞的线程唤醒的呢，又是怎么将节点从等待队列加入到同步队列中的呢？

/** * Moves the longest-waiting thread, if one exists, from the * wait queue for this condition to the wait queue for the * owning lock. * * @throws IllegalMonitorStateException if {@link #isHeldExclusively} *         returns {@code false} */public final void signal() {
   	//当前线程是否占有这个锁，只有持有锁的线程才可以执行唤醒操作    if (!isHeldExclusively())        throw new IllegalMonitorStateException();    Node first = firstWaiter;    if (first != null)    	//从等待队列第一个节点唤醒        doSignal(first);}/** * Removes and transfers nodes until hit non-cancelled one or * null. Split out from signal in part to encourage compilers * to inline the case of no waiters. * @param first (non-null) the first node on condition queue */private void doSignal(Node first) {
       do {
           if ( (firstWaiter = first.nextWaiter) == null)            lastWaiter = null;        //将节点的 nextWaiter 引用清除，方便GC        first.nextWaiter = null;    //如果唤醒节点失败，那么判断节点是否有后续，有则继续循环唤醒节点的后续节点    } while (!transferForSignal(first) &&             (first = firstWaiter) != null);}/** * Transfers a node from a condition queue onto sync queue. * Returns true if successful. * @param node the node * @return true if successfully transferred (else the node was * cancelled before signal) */final boolean transferForSignal(Node node) {
       /*     * If cannot change waitStatus, the node has been cancelled.     */    //节点被取消的话，这里会失败    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))        return false;    /*     * Splice onto queue and try to set waitStatus of predecessor to     * indicate that thread is (probably) waiting. If cancelled or     * attempt to set waitStatus fails, wake up to resync (in which     * case the waitStatus can be transiently and harmlessly wrong).     */    //将节点加入到同步队列的尾节点，不管节点此时是什么状态    Node p = enq(node);    //p 代表的是 node 的前驱    int ws = p.waitStatus;    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))    	//节点前驱被取消，或者设置前驱状态失败（Node.SIGNAL的意义看上篇文章），需要唤醒此节点前驱        LockSupport.unpark(node.thread);    return true;}

signalAll() 方法相当于对等待队列中的每一个节点都执行了一次 signal() 方法

/** * Removes and transfers all nodes. * @param first (non-null) the first node on condition queue */private void doSignalAll(Node first) {
       lastWaiter = firstWaiter = null;    do {
           Node next = first.nextWaiter;        first.nextWaiter = null;        transferForSignal(first);        first = next;    } while (first != null);}

Condition 类分析到这里，举一个例子说明，线程1先获取锁然后等待信号，线程2等待3秒后获取锁后释放一个信号：

/** * Created by Tangwz on 2019/6/21 */public class TestCondition {
       public static void main(String[] args) {
           ReentrantLock reentrantLock = new ReentrantLock(true);        Condition condition = reentrantLock.newCondition();        Thread thread1 = new Thread(new Runnable() {
               @Override            public void run() {
                   reentrantLock.lock();                try {
                       System.out.println("等待一个信号...");                    condition.await();                    System.out.println("拿到一个信号");                } catch (InterruptedException e) {
                       System.out.println("异常");                    e.printStackTrace();                } finally {
                       reentrantLock.unlock();                }            }        });        Thread thread2 = new Thread(() -> {
               try {
                   Thread.sleep(3000);            } catch (InterruptedException e) {
                   e.printStackTrace();            }            reentrantLock.lock();            try {
                   condition.signal();                //想要测试中断放开这里//                thread1.interrupt();//                System.out.println("中断thread1...");                System.out.println("释放一个信号...");            } finally {
                   reentrantLock.unlock();            }        });        thread1.start();        thread2.start();    }}

总结下 Condition 工作流程：

• 线程1调用 reentrantLock.lock() 时，线程被加入到 AQS 的等待队列中，假设竞争锁成功。
• 线程1调用 await 方法时，创建一个等待节点加入到 Condition 的等待队列中，然后释放锁并从 AQS 中移除。
• 线程1判断是否进入了同步队列，没有就阻塞等待 signal 信号。这里假设没有立即被其他线程加入Condition。
• 线程2因为线程1释放锁的关系，被唤醒，并判断可以获取锁，于是线程2获取锁，并被加入到 AQS 的等待队列中。
• 线程2调用 signal() 方法，这个时候 Condition 的等待队列中只有线程1一个节点，于是它被取出来，并被加入到 AQS 的等待队列中。 注意，这个时候，线程1 并没有被唤醒。
• signal方法执行完毕，线程2调用 reentrantLock.unLock() 方法，释放锁。这个时候因为 AQS 中只有线程1，于是，AQS 释放锁后按从头到尾的顺序唤醒线程时，线程1被唤醒，于是线程1恢复执行。
• 线程1判断进入了等待队列，开始获取锁，获取锁成功判断是否被中断，后续操作执行完再释放锁，整个过程执行完毕。

AQS 的共享式同步状态获取、独占式超时获取之后分析读写锁、阻塞队列等再具体分析。

转载地址：http://hmrai.baihongyu.com/