重入锁Condition源码分析

　　参考 https://zhuanlan.zhihu.com/p/81017109 感谢原作者

=================================2020-11-24=============================

突然想到一个问题，那就是ConditionObject其实是在AQS里的，一个线程包成的Node要在同步队列和条件队列里移动，那就得保证是在同一个AQS的实现类里。

JAVA也确实是这么写的

ReentrantLock并没有直接实现AQS，而是通过内部类的方式

public Condition newCondition() {
        return sync.newCondition();
    }

public void lock() {
        sync.lock();
    }

还真的是同一个AQS

　　首先需要明确的是，Condition只工作在排他锁，一个排他锁可以有多个Condition,不过Condition的代码其实是在AQS里的

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;

　　顺便复习一下Node，

    static final class Node {
        /** Marker to indicate a node is waiting in shared mode */
        static final Node SHARED = new Node();
        /** Marker to indicate a node is waiting in exclusive mode */
        static final Node EXCLUSIVE = null;

        /** waitStatus value to indicate thread has cancelled */
        static final int CANCELLED =  1;
        /** waitStatus value to indicate successor's thread needs unparking */
        static final int SIGNAL    = -1;
        /** waitStatus value to indicate thread is waiting on condition */
        static final int CONDITION = -2;
        /**
         * waitStatus value to indicate the next acquireShared should
         * unconditionally propagate
         */
        static final int PROPAGATE = -3;

        volatile int waitStatus;

        volatile Node prev;

        volatile Node next;

        volatile Thread thread;

        Node nextWaiter;

　　上面需要注意的是在同步队列中的时候，是有前驱和后继的prev，next，但是在条件队列中只有nextWaiter，没有前驱，因为没必要

二 condtion.await()

public final void await() throws InterruptedException {
            //判断线程是否中断，如果线程处于中断状态是不能将线程加入condition列队中
            if (Thread.interrupted())
                throw new InterruptedException();
            //将线程封装成node节点加入到condition列队中去
            Node node = addConditionWaiter();
            //释放当前线程拥有的锁资源调用 await 方法时, 当前线程是必须已经获取了独占的锁
            long savedState = fullyRelease(node);
            int interruptMode = 0;
            //判断当前线程是否在 Sync Queue 里面(这里 Node 从 Condtion Queue 里面转移到 Sync Queue 里面有两种可能
            // (1) 其他线程调用 signal 进行转移 
            // (2) 当前线程被中断而进行Node的转移(就在checkInterruptWhileWaiting里面进行转移))
            while (!isOnSyncQueue(node)) {
                //将线程设置成block状态，此时节点处于阻塞状态不在继续向下执行，直到signal()方法唤醒该线程继续向下执行
                LockSupport.park(this);
                //判断此次线程的唤醒是否因为线程被中断, 若是被中断, 则会在checkInterruptWhileWaiting的transferAfterCancelledWait 
                //进行节点的转移; 返回值 interruptMode != 0
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    //// 说明此是通过线程中断的方式进行唤醒, 并且已经进行了 node 的转移, 转移到 Sync Queue 里面
                    break;
            }
            //调用 acquireQueued在 Sync Queue 里面进行 独占锁的获取, 返回值表明在获取的过程中有没有被中断过
            //如果等于THROW_IE证明在等待过程被中断了
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)//即使是被中断的，condition的await也不会立刻返回还是得去竞争锁去，竞争到锁代码才能继续
                interruptMode = REINTERRUPT;
            //通过 "node.nextWaiter != null" 判断 线程的唤醒是中断还是 signal, 因为通过中断唤醒的话, 
            //此刻代表线程的 Node 在 Condition Queue 与 Sync Queue 里面都会存在
            if (node.nextWaiter != null) // clean up if cancelled
                //清除 cancelled 节点
                unlinkCancelledWaiters();
            //"interruptMode != 0" 代表通过中断的方式唤醒线程
            if (interruptMode != 0)
                //根据 interruptMode 的类型决定是抛出异常, 还是自己再中断一下
                reportInterruptAfterWait(interruptMode);
        }

　　上面原作者的注释写的非常好，看了两遍就明白了好多。我再稍微总结下

　　1 执行await后，先加入条件队列，然后才是释放锁，由于新的线程会进行setHeader，所以执行await的线程就会从同步队列中被去掉

　　2 重点要分析中断代码，因为中断也会让park返回，此时把线程从条件队列转移到原同步队列

　　3 如果是正常通过signal结束park的话，会执行竞争锁逻辑，竞争不到阻塞，竞争到了方法才会返回 

private int checkInterruptWhileWaiting(Node node) {
        //判断线程是否中断了，如果中断了调用transferAfterCancelledWait判断中断后是否被清除了
        //线程被清除中断证明node节点处于清除状态后被列队清除掉了
        //当线程等待超时或者被中断，则取消等待，设等待状态为-1，进入取消状态则不再变化

        return Thread.interrupted() ?
             (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
              0;
    }

    final boolean transferAfterCancelledWait(Node node) {
        //通过cas设置node节点状态为0，预期值为condition，是否能设置成功
        //signalled之前发生中断,因为signalled之后会将会将节点状态从CONDITION 设置为0
        if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
            //如果设置成功调用enq方法将node加入到sync列队中去
            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.
         */
        //如果设置失败判断是否在sync列队中，如果没在列队中则将当前线程谦让出去返回false
        // signalled之后发生中断，这个分支的意思就是signal之后才发生的中断，这种算是半正常情况
        // 如果节点还没有被放入同步队列，则放弃当前CPU资源
        // 让其他任务执行，因为此时线程已经中断了而且还没有在sync列队中那么就让出当前cpu资源
        while (!isOnSyncQueue(node))
            Thread.yield();
        return false;
    }

　　 transferAfterCancelledWait 如果返回是false的话，中断类型就是  REINTERRUPT，这种中断类型不需要把异常往上抛

　　有趣的地方：能够看到await在一开始的地方并没有检查当前线程是否正持有锁，而是直到执行到fullyRelease才会触发报错，那么问题来了，当前线程已经包成了Node加入到了条件队列了。

　　如果线程死掉那这个Node怎么办。李老爷子哪能想不到呢，finally里面  node.waitStatus = Node.CANCELLED; 可以保证会在之后该节点清理掉

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

三 signal

public final void signal() {
        //isHeldExclusively判断当前线程是否获取到锁
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    if (first != null)
        //唤醒等待列队中的第一个线程
    doSignal(first);
    }

　　

private void doSignal(Node first) {
            do {
                if ( (firstWaiter = first.nextWaiter) == null)//这种情况说明first后面就没有节点了，同时firstWaiter已经指向了first的next
                    lastWaiter = null;
                first.nextWaiter = null;
            } while (!transferForSignal(first) &&
                     (first = firstWaiter) != null);
        }

　　 先把Node从队列中去掉，也就是 first.nextWaiter = null; 断开条件队列

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);
        int ws = p.waitStatus;
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))//enq的返回值是上一次的tail，这句话的意思是如果tail是取消，或者CAS SIGNAL失败，就唤醒node的线程一次，
　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　//这是防止可能不会唤醒，其实正常流程只需要把节点从条件队列里摘除，同时放入同步队列就可以了
            LockSupport.unpark(node.thread);
        return true;
    }

　　代码量中的不大，英文的注释也说的很明白。先把Node入同步队列，然后要把node前面的节点CAS成SIGNAL,

　　正常的流程到这里就可以了。　　

　　如果失败了就唤醒node的线程一次，再次获取锁的时候，在把自己park之前会检查自己的前驱节点的waitStatus，并且把所有已经取消的节点中队列中去掉

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            /*
             * This node has already set status asking a release
             * to signal it, so it can safely park.
             */
            return true;
        if (ws > 0) {
            /*
             * Predecessor was cancelled. Skip over predecessors and
             * indicate retry.
             */
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            /*
             * waitStatus must be 0 or PROPAGATE.  Indicate that we
             * need a signal, but don't park yet.  Caller will need to
             * retry to make sure it cannot acquire before parking.
             */
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

四 总结

　　await：先把当前线程包成Node加入到条件队列，然后释放锁，这样自然就从同步队列中被去掉了，最后把自己阻塞住。

　　这里会检查中断，如果线程是由于中断被唤醒，就把线程再次加入到同步队列中，但是此时线程已经是取消状态，并且会抛出异常。

　　signal：把队头节点从队头去掉，CAS改变waitStatus为0，把节点加入到同步队列中，正常情况就到此结束了。如果node加入后它的前驱是已取消状态或者CAS成SIGNAL失败，就唤醒一次