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

• AQS结构
• Node结构
• AQS阻塞链表
• ReentrantLock
• ReentrantLock FairSync
• ReentrantLock NonfairSync
• Condition
• CountDownLatch

private transient volatile Node head;
// 阻塞的尾节点，每个新的节点进来，都插入到最后，也就形成了一个隐视的链表
private transient volatile Node tail;
// 这个是最重要的，不过也是最简单的，代表当前锁的状态，0代表没有被占用，大于0代表有线程持有当前锁
// 之所以说大于0，而不是等于1，是因为锁可以重入嘛，每次重入都加上1
private volatile int state;
// 代表当前持有独占锁的线程，举个最重要的使用例子，因为锁可以重入
// reentrantLock.lock()可以嵌套调用多次，所以每次用这个来判断当前线程是否已经拥有了锁
// if (currentThread == getExclusiveOwnerThread()) {state++}
private transient Thread exclusiveOwnerThread; //继承自AbstractOwnableSynchronizer

Node结构

 

//用来表示在共享模式下等待的标记
        static final Node SHARED = new Node();
        
        //用来表示在独占模式下等待的标记
        static final Node EXCLUSIVE = null;

        /** waitStatus 的值，用来表示线程已经被取消 */
        static final int CANCELLED =  1;
        /** waitStatus的值，用来表示后继线程都需要被挂起*/
        static final int SIGNAL    = -1;
        /** waitStatus 的值，表示线程在一个条件上等待 */
        static final int CONDITION = -2;
        /**
         * waitStatus 的值，表示下一个acquireShared应该无条件传播
         */
        static final int PROPAGATE = -3;

        
        volatile int waitStatus;

        //前驱节点
        volatile Node prev;

        //后继节点
        volatile Node next;

        // 当前线程
        volatile Thread thread;

static final class FairSync extends Sync {
      // 争锁
    final void lock() {
        acquire(1);
    }
}

public final void acquire(int arg) {
        //尝试获取锁失败后,这个时候需要把当前线程挂起，放到阻塞队列中;
        //如果在获取锁的等待过程中发生中断，则执行selfInterrupt
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

protected final boolean tryAcquire(int acquires) {
            //获取当前线程
            final Thread current = Thread.currentThread();
            //获取当前节点的状态
            int c = getState();
            //state == 0 此时此刻没有线程持有锁
            if (c == 0) {
                //如果没有前驱节点在排队，则CAS获取锁；如果获得了锁则将独占锁的持有者设置为当前线程
                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;
            }
            //如果没有获取到锁则返回false
            return false;
        }

private Node addWaiter(Node mode) {
    //对当前线程创建一个新节点
    Node node = new Node(Thread.currentThread(), mode);
    //把当前node加到链表的最后面去，也就是进到阻塞队列的最后
    Node pred = tail;
    //如果tail节点不为null
    if (pred != null) {
        //将当前节点的前驱节点设置为目前的队尾节点
        node.prev = pred;
        //CAS将当前节点设置为队尾，如果设置成功则将之前的队尾的后继节点设置为当前节点，并返回
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    //pred==null(队列是空的) 或者 CAS失败(有线程在竞争入队)，则执行enq(node)
    //采用自旋的方式入队
    enq(node);
    return node;
}
// 采用自旋的方式入队
// 之前说过，到这个方法只有两种可能：阻塞队列为空，或者有线程竞争入队，
// 自旋在这边的语义是：CAS设置tail过程中，竞争一次竞争不到，就多次竞争，总会排到的
private Node enq(final Node node) {
    //自旋
    for (;;) {
        //队尾节点
        Node t = tail;
        //如果队尾为空
        if (t == null) { 
             //初始化head节点，原来head和tail初始化的时候都是null，
             //此时初始化阻塞队列，CAS设置一个新节点作为头节点
            if (compareAndSetHead(new Node()))
                //这个时候有了head，但是tail还是null,需要把tail指向head
                tail = head;
        } else {
            //队尾节点不为空，这个套在无限循环里，反正就是将当前线程排到队尾，
            //有线程竞争的话排不上重复排
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}
//下面这个方法，参数node，经过addWaiter(Node.EXCLUSIVE)，此时已经进入阻塞队列
 final boolean acquireQueued(final Node node, int arg) {
     boolean failed = true;
     try {
         //是否中断标记
         boolean interrupted = false;
         //自旋
         for (;;) {
             //获取前驱节点
             final Node p = node.predecessor();
              // p == head 说明当前节点虽然进到了阻塞队列，但是是阻塞队列的第一个，因为它的前驱是head
             // 注意，阻塞队列不包含head节点，head一般指的是占有锁的线程，head后面的才称为阻塞队列；
             //如果前驱节点是头节点则可以执行tryAcquire,因为初始化队列的时候头节点并不占有锁，
             //tryAcquire方法执行成功表示已经抢到锁；则将该节点设置为头节点，后继节点为null
             if (p == head && tryAcquire(arg)) {
                 setHead(node);
                 p.next = null; // help GC
                 failed = false;
                 return interrupted;
             }
             //如果不是头节点或者没有抢到锁则需要判断当前线程是否需要挂起
             if (shouldParkAfterFailedAcquire(p, node) &&
                 parkAndCheckInterrupt())
                 interrupted = true;
         }
     } finally {
         if (failed)
             cancelAcquire(node);
     }
 }

/*
 * 节点获取锁失败后，检查，更新节点waitstatus
 */
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    //获取前驱节点的waitStatus的值
    int ws = pred.waitStatus;
    //前驱节点的 waitStatus == -1 ，说明前驱节点状态正常，当前线程需要挂起，直接可以返回true
    if (ws == Node.SIGNAL)
        return true;
    if (ws > 0) {
        /*
         * 前驱节点 waitStatus大于0 ，大于0 说明前驱节点取消了排队
         * 进入阻塞队列排队的线程会被挂起，而唤醒的操作是由前驱节点完成的。
         * 所以下面这块代码说的是将当前节点的prev指向waitStatus<=0的节点，跳过已经取消的节点
         */
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        /*
         *前驱节点的waitStatus不等于-1和1；都没有看到有设置waitStatus的，
         *所以每个新的node入队时，waitStatu都是
         * 用CAS将前驱节点的waitStatus设置为Node.SIGNAL(也就是-1)
         */
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
        return false;
}

 private final boolean parkAndCheckInterrupt() {
        //当前线程挂起
        LockSupport.park(this);
        //中断检查
        return Thread.interrupted();
    }

  public void unlock() {
        sync.release(1);
    }

   public final boolean release(int arg) {
        //尝试释放锁，如果释放成功
        if (tryRelease(arg)) {
            Node h = head;
            //头节点不为null且waitStatus不是0则唤醒后继节点
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }


    protected final boolean tryRelease(int releases) {
            //当前状态减去releases;
            int c = getState() - releases;
            //判断当前线程是否是持有锁的线程，该方法可以放在第一句执行
            if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
            // 是否完全释放锁
            boolean free = false;
            //如果c=0表示锁已经完全释放了
            if (c == 0) {
                free = true;
                //将持有锁的线程设置为null
                setExclusiveOwnerThread(null);
            }
            //将c的值更新到state
            setState(c);
            return free;
        }


    /**
     *如果存在后继节点则唤醒后继节点
     */
    private void unparkSuccessor(Node node) {
        
        //获取当前节点的值
        int ws = node.waitStatus;
        //如果当前节点的status为负数则CAS修改为0
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);

        //下面的代码就是唤醒后继节点，但是有可能后继节点取消了等待（waitStatus==1）

        //获取当前节点的后继节点
        Node s = node.next;
        //如果后继节点为null或后继节点已经被取消
        if (s == null || s.waitStatus > 0) {
            s = null;
            //从后往前找到最前面未被取消的节点
            for (Node t = tail; t != null && t != node; t = t.prev)
                if (t.waitStatus <= 0)
                    s = t;
        }
        //如果可以找到等待最久的节点就唤醒该节点的线程
        if (s != null)
            LockSupport.unpark(s.thread);
    }

/**
     *  在指定的时间内线程没有中断则获取没有被其他线程持有的锁；如果竞争到锁则立即返回true,
     *
     **/
    public boolean tryLock(long timeout, TimeUnit unit)
            throws InterruptedException {
        return sync.tryAcquireNanos(1, unit.toNanos(timeout));
    }

/**
  * 尝试在独占模式下获取锁，线程中断活着超时情况下终止。
  **/
  public final boolean tryAcquireNanos(int arg, long nanosTimeout)
            throws InterruptedException {
          //线程中断检查
        if (Thread.interrupted())
            throw new InterruptedException();
            //首先尝试获取锁，如果失败，则一直尝试获取锁为止，或超时以及线程中断
        return tryAcquire(arg) ||
            doAcquireNanos(arg, nanosTimeout);
    }


    private boolean doAcquireNanos(int arg, long nanosTimeout)
            throws InterruptedException {
         //如果指定的超时时间小于或等于0则直接返回false       
        if (nanosTimeout <= 0L)
            return false;
        //根据当前时间加上指定的超时时间得到超时时间点
        final long deadline = System.nanoTime() + nanosTimeout;
        //向阻塞队列中添加节点
        final Node node = addWaiter(Node.EXCLUSIVE);
        boolean failed = true;
        try {
            //自旋
            for (;;) {
                //获取当前节点的直接前驱节点 
                final Node p = node.predecessor();
                //如果直接前驱是head节点则尝试获取锁，如果可以获取锁，则将该节点设置为头节点，同时将该节点的后继节点设置为null
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return true;
                }
                //计算剩余等待时间
                nanosTimeout = deadline - System.nanoTime();
                //如果剩余可等待时间小于或等于0则返回false
                if (nanosTimeout <= 0L)
                    return false;
                //还可以等待一段时间，则检查并更新线程状态，如果需要被挂起且剩余等待时间大于自旋优先的阈值，则将该线程最多挂起nanosTimeout 纳秒
                if (shouldParkAfterFailedAcquire(p, node) &&
                    nanosTimeout > spinForTimeoutThreshold)
                    //最多挂起 nanosTimeout 纳秒
                    LockSupport.parkNanos(this, nanosTimeout);
                if (Thread.interrupted())
                    throw new InterruptedException();
            }
        } finally {
            //如果超时或线程中断抛出InterruptedException异常时failed为true
            if (failed)
                cancelAcquire(node);
        }
    }

    /**
     * 取消正在获取锁的尝试
     */
    private void cancelAcquire(Node node) {
        // 如果节点不存在则直接返回
        if (node == null)
            return;
        //将node节点持有的线程设置为null
        node.thread = null;

        // 跳过被取消的前驱节点
        Node pred = node.prev;
        while (pred.waitStatus > 0)
            node.prev = pred = pred.prev;

        // predNext is the apparent node to unsplice. CASes below will
        // fail if not, in which case, we lost race vs another cancel
        // or signal, so no further action is necessary.
        //前驱节点的后继节点
        Node predNext = pred.next;

        //将当前节点的等待状态设置为取消状态
        node.waitStatus = Node.CANCELLED;

        // 如果当前节点就是队尾，将该节点删除
        if (node == tail && compareAndSetTail(node, pred)) {
            compareAndSetNext(pred, predNext, null);
        } else {
            //如果自己不是队尾节点
            int ws;
            //前驱节点也不是头节点且 等待状态为-1或cas更新为-1同时前驱节点线程不为null
            if (pred != head &&
                ((ws = pred.waitStatus) == Node.SIGNAL ||
                 (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
                pred.thread != null) {
                //当前节点的后继节点
                Node next = node.next;
                  //后继节点不为null且没有持有锁，则将前驱节点的直接后继执行后继节点，进而将该节点从链表中删除
                if (next != null && next.waitStatus <= 0)
                    compareAndSetNext(pred, predNext, next);
            } else {
                //如果前驱节点不满足上述条件则唤醒前驱节点
                unparkSuccessor(node);
            }
            //将该节点的后继执行自己，断开链接
            node.next = node; // help GC
        }
    }

/**
     * 竞争独占锁，线程中断则终止
     */
    public final void acquireInterruptibly(int arg)
            throws InterruptedException {
            //首先检查线程是否中断，如果中断则抛出异常
        if (Thread.interrupted())
            throw new InterruptedException();
          //尝试获取锁，如果获取锁失败则执行doAcquireInterruptibly
        if (!tryAcquire(arg))
            doAcquireInterruptibly(arg);
    }

       private void doAcquireInterruptibly(int arg)
        throws InterruptedException {
         //将当前节点添加到阻塞队列   
        final Node node = addWaiter(Node.EXCLUSIVE);
        boolean failed = true;
        try {
            //自旋
            for (;;) {
                //获取当前节点的直接前驱节点
                final Node p = node.predecessor();
                //如果直接前驱是head节点则尝试获取锁，如果可以竞争到锁则将当前节点设置为头节点
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
                //判断需要挂起，同时线程已经中断则抛出中断异常
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
            // 如果通过 InterruptedException 异常出去，那么 failed 就是 true 了
            if (failed)
                cancelAcquire(node);
        }
    }