AQS源码分析(通过ReentrantLock，CountDownLatch分析源码)（下篇）

上面主要讲述了公平锁，如果是非公平锁又是怎样的呢？

NonfairSync

static final class NonfairSync extends Sync {
        private static final long serialVersionUID = 7316153563782823691L;

        
        final void lock() {
           //直接进行CAS尝试进行获取锁，这个比公平锁多出来的操作
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
                acquire(1);
        }

        protected final boolean tryAcquire(int acquires) {
            return nonfairTryAcquire(acquires);
        }
    }


    final boolean nonfairTryAcquire(int acquires) {
            //获取当前线程
            final Thread current = Thread.currentThread();
            //获取state值
            int c = getState();
            //c为0，表示没有线程持有锁，则进行cas操作获取锁
            if (c == 0) {
                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;
        }

Condition

     上面主要走读了独占锁的实现逻辑，下面主要走读Condition相关的实现逻辑

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 obj) throws InterruptedException {
        lock.lock();
        try {
            //如果当前数量已经达到上限则在notFull条件上等待
            while (count == items.length) {
                notFull.await();
            }
            items[putptr] = obj;
            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 obj = items[takeptr];
            if (++takeptr == items.length) takeptr = 0;
            --count;
            notFull.signal();
            return obj;
        } finally {
            lock.unlock();
        }
    }
}

 // 条件队列的第一个节点
        private transient Node firstWaiter;
        // 条件队列的最后一个节点
        private transient Node lastWaiter;

 首先看在条件上阻塞操作

public final void await() throws InterruptedException {
            //如果线程已经中断，则抛出中断异常
            if (Thread.interrupted())
                throw new InterruptedException();
            //向条件等待队列中添加一个新的Waiter节点
            Node node = addConditionWaiter();
            //完全释放该节点的锁，只有完全释放才可以避免重入问题，并将释放前的值返回；后续再次获取锁的时候需要用到
            int savedState = fullyRelease(node);
            int interruptMode = 0;
            //如果当前节点已经在阻塞队列中或者在等待过程中中断过，如果该节点不在阻塞队列中则线程在该处挂起

            while (!isOnSyncQueue(node)) {
                LockSupport.park(this);
                //线程唤醒后会进行中断检查
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
            }
            
            //线程唤醒后将该节点加入到阻塞队列  如果在唤醒前就发生了中断则interruptMode 设置为REINTERRUPT
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            //如果后继节点不为null 则进行清理取消的节点
            if (node.nextWaiter != null) // clean up if cancelled
                unlinkCancelledWaiters();
                //interruptMode不为0则会根据interruptMode的值决定是抛出异常还是执行中断或则什么都不做
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
        }

     将节点添加到等待队列

       /**
         * 向等待队列中添加一个新的waiter,并返回这个新的等待节点
         */
        private Node addConditionWaiter() {
            //等待队列的最后一个节点
            Node t = lastWaiter;
            // If lastWaiter is cancelled, clean out.
            //如果最后一个节点不为null但是状态又不是condition,则需要清理等待队列
            if (t != null && t.waitStatus != Node.CONDITION) {
                unlinkCancelledWaiters();
                t = lastWaiter;
            }
            //新建一个等待节点
            Node node = new Node(Thread.currentThread(), Node.CONDITION);
            //如果最后一个节点为null,则将该节点设置为最后节点，否则将该节点添加到最后节点之后并将该节点设置为最后节点
            if (t == null)
                firstWaiter = node;
            else
                t.nextWaiter = node;
            lastWaiter = node;
            return node;
        }

//等待队列是一个单向链表，遍历链表将已经取消等待的节点清除出去
        private void unlinkCancelledWaiters() {
            //获取第一个等待节点
            Node t = firstWaiter;
            Node trail = null;
            //如果第一个节点不为null
            while (t != null) {
                //获取下一个节点
                Node next = t.nextWaiter;
                //如果第一个节点的等待状态不是condition
                if (t.waitStatus != Node.CONDITION) {
                    //将下一个节点的直接后继设置为null
                    t.nextWaiter = null;
                    //如果trail为null则将第二个节点提升为第一个节点，否则trail的直接后继节点指向next,将当前节点从链表中删除
                    if (trail == null)
                        firstWaiter = next;
                    else
                        trail.nextWaiter = next;
                    //遍历到队尾，将trial指向到节点设置为最后一个等待节点
                    if (next == null)
                        lastWaiter = trail;
                }
                else//如果第一个等待节点是condition状态，则trail指向第一个节点，t指向下一个节点
                    trail = t;
                t = next;
            }
        }

final int fullyRelease(Node node) {
        boolean failed = true;
        try {
            //获取当前同同步状态
            int savedState = getState();
            //这里使用了当前的 state 作为 release 的参数，也就是完全释放掉锁，将 state 置为 0
            if (release(savedState)) {
                failed = false;
                return savedState;
            } else {
                throw new IllegalMonitorStateException();
            }
        } finally {
            if (failed)
                node.waitStatus = Node.CANCELLED;
        }
    }

final boolean isOnSyncQueue(Node node) {
        //如果 waitStatus 还是 Node.CONDITION，也就是 -2，那肯定就是还在条件队列中
        //如果 node 的前驱 prev 指向还是 null，说明肯定没有在 阻塞队列
        if (node.waitStatus == Node.CONDITION || node.prev == null)
            return false;
        //如果存在后继节点则一定在阻塞队列中
        //node.prev() != null 来推断出 node 在阻塞队列？不能是因为在addWaiter方法中会将node.prev指向为队尾，但是此时
        //节点上位加入到阻塞队列中
        if (node.next != null) 
            return true;
        // 这个方法从阻塞队列的队尾开始从后往前遍历找，如果找到相等的，说明在阻塞队列，否则就是不在阻塞队列
        return findNodeFromTail(node);
    }

   
    private boolean findNodeFromTail(Node node) {
        Node t = tail;
        for (;;) {
            //如果t==node表示该节点在阻塞队列中
            if (t == node)
                return true;
            //如果全部遍历完还是没有找到相同的节点则返回false
            if (t == null)
                return false;
            t = t.prev;
        }
    }

  如果当前等待节点不在阻塞队列中，则需要执行 LockSupport.park(this)将线程挂起；那么什么时候唤醒呢？后面会详细说；假设此时线程被唤醒，唤醒后需要进行中断检查，如果中断检查当返回值不是0，即发生了中断则从while循环中跳出。

       /**
         * 中断检查，如果在唤醒前中断则返回 THOW_IE,如果在唤醒后中断则返回
         * REINTERRUPT；如果没有中断则返回0 
         */
        private int checkInterruptWhileWaiting(Node node) {
            return Thread.interrupted() ?
                (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
                0;
        }

从while循环中跳出后，执行   if (acquireQueued(node, savedState) && interruptMode != THROW_IE)，acquireQueued是做什么的呢？上文已经有介绍过了，主要是用来锁竞争和线程挂起。如果竞争到了锁且未发生中断，则await阻塞结束，执行后续业务代码。但是我们知道acquireQueued要求当前Node是在阻塞队列的，那么什么时候加入阻塞队列的呢？下面我们看一下signal方法

/**
         * 如果等待队列存在线程，将等待时间最长的线程从等待队列转移到锁的阻塞队列中
         *
         */
        public final void signal() {
            //没有独占锁抛出异常
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            //condition等待队列的第一个节点
            Node first = firstWaiter;
            //如果第一个节点不为null
            if (first != null)
                doSignal(first);
        }

         private void doSignal(Node first) {
            do {
                //当前节点的下一个节点设置为第一个节点，如果这个节点为null则将最后一个节点设置为null，当前节点的下一个节点也设置为null
                if ( (firstWaiter = first.nextWaiter) == null)
                    lastWaiter = null;
                first.nextWaiter = null;
                //循环推出的条件是：将condition等待队列的第一个节点转移到阻塞队列成功或等待队列中元素为空
            } while (!transferForSignal(first) &&
                     (first = firstWaiter) != null);
        }

final boolean transferForSignal(Node node) {
        /*
         *CAS 设置节点waitStatus的值为0，如果设置失败,即当前节点的状态不是CONDITION,标示该节点已经被取消了
         */
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            return false;

     
        //将该节点自旋加入阻塞队列的队尾，并返回该节点的前驱节点
        Node p = enq(node);
        //获取前驱节点的状态
        int ws = p.waitStatus;
        // ws > 0 说明 node 在阻塞队列中的前驱节点取消了等待锁，直接唤醒 node 对应的线程
        //如果 ws <= 0, 那么 compareAndSetWaitStatus 将会被调用，节点入队后，需要把前驱节点的状态设为 Node.SIGNAL(-1)，如果设置失败则唤起线程
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
            LockSupport.unpark(node.thread);
        return true;
    }

如果发生了中断会怎么处理内？

       /**
         * 如果interruptMode是 THROW_IE 则抛出中断异常
         * 如果interruptMode是 REINTERUPT 则再次执行中断
         * 其他情况什么都不做
         */
        private void reportInterruptAfterWait(int interruptMode)
            throws InterruptedException {
            if (interruptMode == THROW_IE)
                throw new InterruptedException();
            else if (interruptMode == REINTERRUPT)
                selfInterrupt();
        }

    至此已经将排它锁相关逻辑介绍完成，下面介绍共享锁相关逻辑。   

CountDownLatch

      CountDownLatch结构

首先看一下await()方法

   public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }

    /**
     * 在共享模式下获取锁，线程中断则终止
     */
    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        //首先进行线程中断检查，如果中断则抛出中断异常
        if (Thread.interrupted())
            throw new InterruptedException();
        //尝试获取锁，如果获取锁失败则执行doAcquireSharedInterruptibly()方法，
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }
    
     //只有state为0的时候才返回1，否则就返回-1
     protected int tryAcquireShared(int acquires) {
            return (getState() == 0) ? 1 : -1;
        }

    /**
     * 在共享可中断模式下获取锁
     */
    private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
        //为当前线程创建一个共享模式节点并加入到阻塞队列中
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            //自旋
            for (;;) {
                //获取当前新建节点的直接前驱节点
                final Node p = node.predecessor();
                //如果直接前驱节点为head节点 
                if (p == head) {
                    //尝试获取锁
                    int r = tryAcquireShared(arg);
                    //获取到锁
                    if (r >= 0) {
                        //将当前节点设置为头节点和传播
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                }
                //如果当前节点的直接前驱不是head或没有获取到锁，则需要判断当前节点是否需要挂起，如果需要则通过parkAndCheckInterrput()方法挂起
                //如果发生中断则抛出异常
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }
    
    /**
     * 
     * 设置队列头，并检查后继节点是否可以在共享模式下等待，如果不需要等待则可以唤醒后继节点
     */
    private void setHeadAndPropagate(Node node, int propagate) {
        //当前头节点
        Node h = head; 
        //将当前节点设置为新的头节点
        setHead(node);
        //如果propagate大于0，通过上面代码可以知道propagate的值为1,如果下一个节点不为null且只共享模式则执行doReleaseShared()方法
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) {
            Node s = node.next;
            if (s == null || s.isShared())
                doReleaseShared();
        }
    }

   /**

     * 共享模式下释放锁的操作,唤醒后继节点同时保证传播功能
     */
    private void doReleaseShared() {
        
        for (;;) {
            //头节点
            Node h = head;
            //如果头节点不为null同时也不能与尾节点（初始化的时候头节点等于尾节点）
            if (h != null && h != tail) {
                //头节点的等待状态
                int ws = h.waitStatus;
                //如果头节点的状态时SINGLE(-1),则将头节点设置为0，失败重试,成功则唤醒头节点
                if (ws == Node.SIGNAL) {
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;          
                    unparkSuccessor(h);
                }//如果头节点是新加入的节点，则将其状态设置为-3
                //为什么不可以是-1呢？这是因为这个方法执行结束后需要执行chouldParkAfterFail,-1在此会被挂起
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;               
            }
            //如果头节点没有变化则跳出循环，反之则继续循环
            if (h == head)                   
                break;
        }
    }  
   

 public void countDown() {
        sync.releaseShared(1);
    }

 public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }

 protected boolean tryReleaseShared(int releases) {
            // 对state值进行自减，如果结果为0则返回true
            for (;;) {
                int c = getState();
                if (c == 0)
                    return false;
                int nextc = c-1;
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        } 

public boolean await(long timeout, TimeUnit unit)
        throws InterruptedException {
        return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
    }

 public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
            //如果没有获取锁则执行doAcquireSharedNanos方法
        return tryAcquireShared(arg) >= 0 ||
            doAcquireSharedNanos(arg, nanosTimeout);
    }


private boolean doAcquireSharedNanos(int arg, long nanosTimeout)
            throws InterruptedException {
        if (nanosTimeout <= 0L)
            return false;
        //计算出deadLine
        final long deadline = System.nanoTime() + nanosTimeout;
        //为当前线程创建一个共享模式节点并添加到阻塞队列
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            for (;;) {
                final Node p = node.predecessor();
                if (p == head) {
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        failed = false;
                        return true;
                    }
                }
                //计算剩下等待时间
                nanosTimeout = deadline - System.nanoTime();
                //如果剩下等待时间小于或等于0则表示超时了，返回false
                if (nanosTimeout <= 0L)
                    return false;
                //判断是否需要挂起，如果剩余等待时间小于阈值则不挂起而是通过自旋处理；
                if (shouldParkAfterFailedAcquire(p, node) &&
                    nanosTimeout > spinForTimeoutThreshold)
                    LockSupport.parkNanos(this, nanosTimeout);
                if (Thread.interrupted())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }