JUC AbstractQueuedSynchronizer分析

清疚 2022-05-23 13:07 181阅读 0赞

## 什么是AQS ##

AQS即`AbstractQueuedSynchronizer`的简称，是实现JUC(java.utils.concurrent包)同步组件的通用基础骨架，JDK中基于此骨架实现的同步组件主要包括`ReentrantLock`、`ReentrantReadWriteLock`、`CountDownLatch`和`Semaphore`。理解好这个基础骨架实现的功能，就能够更好地去理解其他组件的实现。

AQS通过内置的同步状态，以及内置的FIFO同步队列，加以CAS操作的支持，来实现通用的同步功能。另外依靠内置的Condition实现（实现类为ConditionObject，内部维护了FIFO等待队列），来实现线程间通讯——等待和唤醒。

粗略的看，AQS是synchronized和Object.wait和notify的结合体，事实上AQS提供的功能远远强大于synchronized和原始的wait、notify机制，比如说，用AQS实现的锁，支持超时退出和中断，这是原先synchronized无法比拟的。

## 先说说CAS ##

CAS，即*Compare And Set*，能够以原子的方式实现这样的更新:

if (value == expect) { // 值符合预期
        value = update; // 则更新为新的值
    }

> 在Java中，CAS操作的实现封装在底层，而实际依赖的是具体CPU支持的CAS指令，不同架构CPU基于不同的方式和指令来实现CAS，没有了解那么深，无法展开讨论。

AQS内部维护了同步状态（一个定义为volatile的int字段），需要依靠CAS操作来保证原子更新，提供的方法是:

*  compareAndSetState(int expect, int update)

同时，为了能够在多线程并发访问下得以维护内置的FIFO同步队列和ConditionObject中的等待队列，也提供了CAS操作，主要包括:

*  compareAndSetHead(Node update)
 *  compareAndSetTail(Node expect, Node update)
 *  compareAndSetNext(Node node, Node expect, Node update)
 *  compareAndSetWaitStatus(Node node, int expect, int update)

> compareAndSet的具体实现依赖Unsafe这个类，这里不展开；volatile的具体介绍可以参考: [https://www.cnblogs.com/dolphin0520/p/3920373.html][https_www.cnblogs.com_dolphin0520_p_3920373.html]

## AQS如何使用 ##

AQS的目的是用来构建同步组件的，推荐的使用方法就是将其运用到同步组件的私有静态内部类，通过外部类公开的接口，调用内部AQS实现类，来实现具体的同步语义。

> 一般来说JUC中基于AQS实现的同步组件已经能满足大多数开发需求了，自己通过扩展AQS来实现特殊同步组件的需求本身比较少

如下是一个不支持重入的独占锁的实现:

/** * 独占锁的实现(不支持重入) * @author chenjianxin * */
    public class Mutex implements Lock { 
    
        private Sync sync = new Sync();
    
        private static class Sync extends AbstractQueuedSynchronizer { 
    
            /** * */
            private static final long serialVersionUID = 1L;
    
            public Sync() {
    
            }
    
            @Override
            protected boolean tryAcquire(long arg) {
                if (compareAndSetState(0, 1)) {
                    setExclusiveOwnerThread(Thread.currentThread());
                    return true;
                }
                return false;
            }
    
            @Override
            protected boolean tryRelease(long arg) {
                Thread owner = getExclusiveOwnerThread();
                if (owner != Thread.currentThread()) {
                    throw new IllegalMonitorStateException();
                }
                setExclusiveOwnerThread(null);
                setState(0);
                return true;
            }
    
            @Override
            protected boolean isHeldExclusively() {
                return getExclusiveOwnerThread() == Thread.currentThread();
            }
    
            public Condition newCondition() {
                return new ConditionObject();
            }
    
        }
    
        public void lock() {
            sync.acquire(1);
        }
    
        public void lockInterruptibly() throws InterruptedException {
            sync.acquireInterruptibly(1);
        }
    
        public boolean tryLock() {
            return sync.tryAcquire(1);
        }
    
        public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
            return sync.tryAcquireNanos(1, unit.toNanos(time));
        }
    
        public void unlock() {
            sync.release(1);
        }
    
        public Condition newCondition() {
            return sync.newCondition();
        }
    
    }

基于静态内部类实现，你只需要实现你需要的模式即可，因为AQS可同时支持独占式同步实现和共享式的同步实现，而我们的同步组件实现一般来说只会关注其中的一个实现。

## 主要方法 ##

AQS提供了一系列模板方法，通过这些方法来构造通用的同步实现，模板方法如下说明:

<table> 
 <thead> 
  <tr> 
   <th align="left">方法定义</th> 
   <th align="left">方法说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">void acquire(int arg)</td> 
   <td align="left">独占式获取同步状态，如果当前线程获取同步状态成功，则由该方法返回，否则，将会进入同步队列等待，该方法会调用重写的tryAcquire(int arg)方法</td> 
  </tr> 
  <tr> 
   <td align="left">void acquireInterruptibly(int arg)</td> 
   <td align="left">与acquire(int arg)相同，但是该方法支持中断，当前线程未获取到同步状态而进入同步队列中，如果当前线程被中断，则该方法会抛出InterruptedException并返回</td> 
  </tr> 
  <tr> 
   <td align="left">boolean tryAcquireNanos(int arg, long nanos)</td> 
   <td align="left">acquireInterruptibly(int arg)的基础上增加了超时限制，如果当前线程在超时时间内没有获取到同步状态，则会返回false，如果获取到则返回true</td> 
  </tr> 
  <tr> 
   <td align="left">void acquireShared(int arg)</td> 
   <td align="left">共享式获取同步状态，如果当前线程未获取同步状态，将会进入同步队列等待，与独占式获取的主要区别是在同一时刻可以有多个线程获取到同步状态</td> 
  </tr> 
  <tr> 
   <td align="left">void acquireSharedInterruptibly(int arg)</td> 
   <td align="left">在acquireShared(int arg)的基础上支持中断</td> 
  </tr> 
  <tr> 
   <td align="left">boolean tryAcquireSharedNanos(int arg, long nanosTimeout)</td> 
   <td align="left">在acquireSharedInterruptibly(int arg)的基础上增加超时机制，超时机制与独占式的超时实现一致</td> 
  </tr> 
  <tr> 
   <td align="left">boolan release(int arg)</td> 
   <td align="left">独占式释放同步状态，必须由持有同步状态的线程执行，该方法会在释放同步状态之后，将同步队列中下一个节点包含的线程唤醒</td> 
  </tr> 
  <tr> 
   <td align="left">boolean releaseShared(int arg)</td> 
   <td align="left">共享式释放同步状态</td> 
  </tr> 
 </tbody> 
</table>

而模板方法里面，会调用需要我们实现（重写）的方法

<table> 
 <thead> 
  <tr> 
   <th align="left">方法定义</th> 
   <th align="left">方法说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">protected boolean tryAcquire(int arg)</td> 
   <td align="left">独占式获取同步状态，实现该方法需要查询当前状态并判断同步状态是否符合预期，然后再进行CAS设置同步状态</td> 
  </tr> 
  <tr> 
   <td align="left">protected boolean tryRelease(int arg)</td> 
   <td align="left">独占式释放同步状态，等待获取同步状态的线程将有机会获取同步状态</td> 
  </tr> 
  <tr> 
   <td align="left">protected int tryAcquireShared(int arg)</td> 
   <td align="left">共享式获取同步状态，返回&gt;=0的值，表示获取成功，否则，获取失败</td> 
  </tr> 
  <tr> 
   <td align="left">protected boolean tryReleaseShared(int arg)</td> 
   <td align="left">共享式释放同步状态，等待获取同步状态的多个线程将有机会获取同步状态</td> 
  </tr> 
  <tr> 
   <td align="left">protected boolean isHeldExclusively()</td> 
   <td align="left">当前同步器是否在独占模式下被线程占用，一般该方法表示是否能被当前线程独占</td> 
  </tr> 
 </tbody> 
</table>

## 同步队列 ##

AQS依赖内部实现的FIFO同步队列来完成多线程的同步管理。当线程获取同步状态失败，在AQS内部将会把该线程以及一些状态构造成一个节点(Node)，然后将其加入到同步队列的尾部，同时阻塞当前线程，直到获取同步状态的线程释放同步状态，（针对独占式的同步来说）同步队列的头结点的第一个有效后继节点所在的线程会被唤醒，进而继续尝试获取同步状态。

同步队列是一个双端队列，队列的元素构成是AQS中的Node内部类。

### Node ###

AQS的内部类，代表同步队列中的节点（实际上也用来构造等待队列的节点，后面讲到等待队列的时候会提及），主要属性如下所示:

static final class Node {
    
        volatile int waitStatus;
    
        volatile Node prev;
    
        volatile Node next;
    
        volatile Thread thread;
    
        Node nextWaiter;
    
    }

*  waitStatus: 当前节点等待状态，用于控制节点线程的等待与唤醒，取值和说明如下:

<table> 
 <thead> 
  <tr> 
   <th align="left">取值</th> 
   <th align="left">说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">-1</td> 
   <td align="left">SIGNAL: 代表后继节点所在线程处于等待状态，当前节点所在线程释放同步状态或者被取消、中断，需要通知后继节点，使得后继节点再次尝试获取同步状态（从阻塞中唤醒）</td> 
  </tr> 
  <tr> 
   <td align="left">-2</td> 
   <td align="left">CONDITION: 节点在等待队列中，节点所在线程等待在Condition上，当其他线程调用Condition的signal方法后，该节点将会从等待队列中移除并加到同步队列尾部，等待获取同步状态</td> 
  </tr> 
  <tr> 
   <td align="left">-3</td> 
   <td align="left">PROPAGATE: 表示下一次共享式获取同步状态将会无条件地被传播下去</td> 
  </tr> 
  <tr> 
   <td align="left">1</td> 
   <td align="left">CANCEL，由于在同步队列中等待的节点所在线程等待超时或被中断，需要从同步队列中取消等待</td> 
  </tr> 
  <tr> 
   <td align="left">0</td> 
   <td align="left">初始化状态</td> 
  </tr> 
 </tbody> 
</table>

\- prev: 当前同步队列中的上一个节点（前继节点）  
\- next: 当前同步队列中的下一个节点（后继节点）  
\- thread: 当前节点所维护的线程（等待获取同步状态的线程 or 等待Condition signal的线程）  
\- nextWaiter: 指向等待队列的下一个节点（只有在等待队列中才有效的字段）

同步队列如下图所示:

![AQS\_sync\_queue][AQS_sync_queue]

## LockSupport ##

AQS中，当线程无法获取同步状态，则进入线程阻塞，实际上是调用了`LockSupport`的park方法来完成的线程阻塞，对于线程阻塞的解除，依赖于`LockSupport`的unpark方法。这里不展开对`LockSupport`的探讨，我们只要知道通过调用这个类的park和unpark方法，就可以实现线程的阻塞和唤醒。对`LockSupport`更深入的理解，可以参考[https://segmentfault.com/a/1190000008420938][https_segmentfault.com_a_1190000008420938]

## 锁语义的实现分析 ##

通过AQS，可以实现独占锁的语义（如`ReentrantLock`），比`synchronized`更进一步，还可以实现共享锁的语义（如`ReentrantReadWriteLock`），下面我们通过源码来分析，AQS是如何做到实现锁的语义

> synchronized 只能实现独占锁的语义，同一个时刻，只有一个线程能够进入synchronized保护的代码块

### 独占锁语义实现分析 ###

#### 独占锁的获取分析 ####

> 先不考虑支持超时和中断的情况，后续小节再分析支持超时和中断的获取方式

主要入口点在`acquire`方法:

public final void acquire(int arg) {
        // 尝试获取同步状态，如果获取不到，则构造节点加入同步队列尾部，必要时候阻塞线程
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

流程分析:

1）尝试获取同步状态,如果成功，表示可以获取到独占锁，返回  
2）如果获取不到，将构造一个Node，加入同步队列尾部，然后调用acquireQueued方法，还获取不到同步状态，就会在该方法里面让线程阻塞(使用前面说到的`LockSupport.park`)

其中`tryAcquire`是我们的实现类需要覆盖实现的方法，由我们的实现来判断同步状态是否能够被获取。addWaiter方法如下分析:

private Node addWaiter(Node mode) {
        Node node = new Node(Thread.currentThread(), mode);
        // 队列非空(有head节点)，尝试CAS将当前节点加到尾部，若不成功，则会调用enq方法，在enq内部以循环CAS方式将当前节点加到尾部
        Node pred = tail;
        if (pred != null) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        enq(node);
        return node;
    }

enq方法如下所示:

private Node enq(final Node node) {
        // 循环CAS，直到成功将节点加入同步队列尾部
        for (;;) {
            Node t = tail;
            if (t == null) { // 尾节点指向空，证明当前同步队列为空，初始化头节点
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                // CAS尝试将节点加入同步队列尾部
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

最后分析重头戏`acquireQueued`方法:

final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                // 当前节点的前继是头节点，尝试获取同步状态，若成功，将当前节点设置为头节点，然后返回
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                // 上述条件不满足，则判断是否需要阻塞线程，若需要，则当前线程阻塞于parkAndCheckInterrupt方法内(内部调用LockSupport.park)
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

1）进入该方法，会判断当前节点的前继节点是否为头节点，如果是，则调用`tryAcquire`方法尝试获取同步状态，若获取成功，则将当前节点设置为头节点，然后返回。  
2）否则，进入`shouldParkAfterFailedAcquire`方法，判断是否需要将当前节点线程阻塞，如果是，则调用`parkAndCheckInterrupt`方法将线程阻塞，否则，继续循环判断

> 乍一看好像有可能陷入死循环，实际上如果竞争不到同步状态，最终趋向肯定是`shouldParkAfterFailedAcquire`方法返回true，然后调用`parkAndCheckInterrupt`方法让线程阻塞，让出CPU资源

重点分析`shouldParkAfterFailedAcquire`方法:

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            // 当前节点前继节点waitStatus为SIGNAL，则当前节点判定为需要阻塞，返回true，将会使当前节点线程阻塞
            return true;
        if (ws > 0) {
            // 遇到前继节点CANCEL，从后往前找非CANCEL类型的节点
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            // 这里会将所有CANCEL节点从队列中断开，返回false，对于上层调用的acquireQueued方法，继续for循环判断
            pred.next = node;
        } else {
            // waitStatus must be 0 or PROPAGATE
            // 遇到这两种情况，尝试更新前继节点的waitStatus为SIGNAL，返回false，让上层acquireQueued方法继续for循环判断（即使CAS失败）
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

`parkAndCheckInterrupt`方法就没什么好说了，主要就是调用`LockSupport.park`将当前节点线程阻塞，并返回中断状态:

private final boolean parkAndCheckInterrupt() {
        // 阻塞当前线程
        LockSupport.park(this);
        // 返回当前线程是否中断，并重置中断状态为false
        return Thread.interrupted();
    }

经过上面的分析，实际上独占锁语义的最终实现，是取决于我们对于`tryAcquire`的实现，在这个方法里面，切记对于同步状态的更新，需要使用CAS操作，如上面例子，实现一个不支持重入的独占锁所示:

@Override
    protected boolean tryAcquire(long arg) {
        // 同步状态尝试CAS从0更新为1，成功则表示可以获取同步状态，即表示取得独占锁，那么设置当前独占的线程为当前线程，然后返回true
        if (compareAndSetState(0, 1)) {
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        // CAS失败，则返回false，表示无法获取独占锁，根据上面的分析，当前线程最终将会阻塞在AQS的内部实现方法`parkAndCheckInterrupt`
        // 直到获取独占锁的线程调用release方法释放持有的锁，然后由它进行唤醒
        return false;
    }

#### 独占锁的释放分析 ####

需要调用AQS的relase方法进行独占锁的释放:

public final boolean release(int arg) {
        // 返回true，表示成功释放同步状态
        if (tryRelease(arg)) {
            Node h = head;
            // 如果同步队列里面存在等待节点需要唤醒（CANCEL or SIGNAL），则唤醒首节点的下一个非取消的节点线程，让其重新尝试获取同步状态
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

`tryRelease`方法是我们需要进行重写实现的方法，注意只有持有锁的线程才能正确进行释放，所以需要我们在`tryRelease`方法里面进行判断。唤醒操作体现在`unparkSuccessor`方法，下面进行分析:

private void unparkSuccessor(Node node) {
        /* * If status is negative (i.e., possibly needing signal) try * to clear in anticipation of signalling. It is OK if this * fails or if status is changed by waiting thread. */
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);
    
        // 找下一个非CANCEL的后继节点进行唤醒
        Node s = node.next;
        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);
    }

### 共享锁语义实现分析 ###

共享锁允许同一个时刻有多个线程同时访问同步代码，下面分析共享锁语义的实现。

#### 共享锁获取实现分析 ####

同样也是分析非中断和超时的代码，入口在`acquireShared`方法:

public final void acquireShared(int arg) {
        if (tryAcquireShared(arg) < 0)
            doAcquireShared(arg);
    }

通过前面的方法介绍我们知道`tryAcquireShared`方法返回大于等于0时，表示能够获取到共享同步状态，而返回负数，则表示无法获取，那么则进入`doAcquireShared`方法，将会在这个方法里面自旋，阻塞直到被唤醒:

private void doAcquireShared(int arg) {
        // 构造同步队列节点，加入尾部
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                // 与独占式的判断一样，需要判断前继节点是否为首节点
                if (p == head) {
                    int r = tryAcquireShared(arg);
                    // 能够获取共享同步状态
                    if (r >= 0) {
                        // 设置当前节点为头节点并进行传播(让其他线程能够尝试获取同步状态)，并返回
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        if (interrupted)
                            selfInterrupt();
                        failed = false;
                        return;
                    }
                }
                // 到这里则说明需要判断是否需要阻塞当前节点线程，与独占锁的分析流程一致
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

与独占锁语义实现主要不同体现在`setHeadAndPropagate`方法，在共享式获取里面，需要判断是否需要唤醒后继节点:

private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head; // Record old head for check below
        setHead(node);
        /* * Try to signal next queued node if: * Propagation was indicated by caller, * or was recorded (as h.waitStatus either before * or after setHead) by a previous operation * (note: this uses sign-check of waitStatus because * PROPAGATE status may transition to SIGNAL.) * and * The next node is waiting in shared mode, * or we don't know, because it appears null * * The conservatism in both of these checks may cause * unnecessary wake-ups, but only when there are multiple * racing acquires/releases, so most need signals now or soon * anyway. */
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) {
            Node s = node.next;
            // 只有节点为共享模式下，才唤醒???
            if (s == null || s.isShared())
                doReleaseShared();
        }
    }

`propagate > 0`，证明仍有线程可以尝试获取同步状态，此时需要唤醒后继节点。而`h.waitStatus < 0`，表明当前节点状态可能为 SIGNAL，CONDITION，PROPAGATE，同样需要尝试唤醒，让其尝试获取同步状态。

注意在`doReleaseShared`方法里面，只处理唤醒，并不修改同步状态:

private void doReleaseShared() {
        /* * Ensure that a release propagates, even if there are other * in-progress acquires/releases. This proceeds in the usual * way of trying to unparkSuccessor of head if it needs * signal. But if it does not, status is set to PROPAGATE to * ensure that upon release, propagation continues. * Additionally, we must loop in case a new node is added * while we are doing this. Also, unlike other uses of * unparkSuccessor, we need to know if CAS to reset status * fails, if so rechecking. */
        for (;;) {
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                if (ws == Node.SIGNAL) {
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
                    unparkSuccessor(h);
                }
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            if (h == head)                   // loop if head changed
                break;
        }
    }

#### 共享锁释放实现分析 ####

入口方法`releaseShared`:

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

其中`tryReleaseShared`方法需要我们重写实现，注意，在这里面一般使用循环CAS修改同步状态，并且注意，当返回`>=0`，表示能够成功获取同步状态，对于`>0`的情况，表示其他线程还有机会同时获取同步资源。而返回`<0`，则表示无法获取同步状态。`doReleaseShared`方法在上面已经分析，不再赘述。

## 锁的中断和锁超时分析 ##

使用AQS能够构造支持中断的锁，无论独占锁还是共享锁。这是AQS实现的锁区别于JVM内置锁(synchronized)的又一大优势。这里拿独占式获取来分析

### 锁中断实现分析 ###

入口方法是:

public final void acquireInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        if (!tryAcquire(arg))
            doAcquireInterruptibly(arg);
    }

首先也是调用`tryAcquire`方法尝试获取同步状态，失败则调用`doAcquireInterruptibly`:

private void doAcquireInterruptibly(int arg)
        throws InterruptedException {
        final Node node = addWaiter(Node.EXCLUSIVE);
        boolean failed = true;
        try {
            for (;;) {
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

看到这里，实际上发现跟`acquireQueued`方法几乎是一致的。实现中断的关键在于: 当线程阻塞于`parkAndCheckInterrupt`方法，遇到中断，则会调用`throw new InterruptedException()`，往上层调用抛`InterruptedException`。

### 锁超时实现分析 ###

使用AQS实现的锁，还可以实现超时获取，如果在约定的时间内没有获取到锁，认定为超时，会退出阻塞状态。相比synchronized，这是又一大优势。同时注意，支持超时的同时也支持中断。

> 在一定的场合下，使用超时获取锁，可以有效避免死锁的发生。

下面我们来看看如何实现的超时机制，同样，拿独占式获取来说明，入口方法是`tryAcquireNanos`

public final boolean tryAcquireNanos(int arg, long nanosTimeout)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        return tryAcquire(arg) ||
            doAcquireNanos(arg, nanosTimeout);
    }

大同小异，主要分析`doAcquireNanos`:

private boolean doAcquireNanos(int arg, long nanosTimeout)
            throws InterruptedException {
        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();
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return true;
                }
                nanosTimeout = deadline - System.nanoTime();
                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);
        }
    }

可以发现，实现超时获取的关键是`LockSupport`的`parkNanos`方法。如果阻塞线程超过指定的时间，该方法返回，同时，因为`failed = true`，则会调用finally块的`cancelAcquire`方法，取消当前节点线程对同步状态的获取，将该节点从同步队列中摘除。

这里注意到，如果`nanosTimeout < spinForTimeoutThreshold`，证明距离超时时间太短，无需经过park，直接判断为超时。

## Condition分析 ##

从前面的分析可知，AQS可用于实现锁语义，包括独占锁和共享锁，同时支持中断获取锁以及超时获取锁，除此之外，AQS还通过内置的Condition实现，来实现线程间的通讯——等待与通知，可类比为Object的wait和notify。

原始的Object wait和notfiy相当于提供了单对象单Condition的功能，而对于一个AQS的实现，可能同时具备多个Condition，相当于可实现单对象（指AQS的实现）多Condition的功能。

**有一点需要注意，只有实现为独占锁的AQS，才能够实现Condition的等待和唤醒，共享锁无法实现Condition。**

AQS通过内部类`ConditionObject`实现了Condition接口，ConditionObject内部维护了一个FIFO等待队列，用于实现线程间等待通知机制。更进一步，AQS可以同时维护多个等待队列，从而实现多Condition的功能。

![AQS\_wait\_queue][AQS_wait_queue]

### ConditionObject与Condition ###

AQS通过内部类ConditionObject实现了Condition接口，Condition接口定义了需要实现的等待和通知接口:

public interface Condition { 
    
        void await() throws InterruptedException;
    
        void awaitUninterruptibly();
    
        long awaitNanos(long nanosTimeout) throws InterruptedException;
    
        boolean await(long time, TimeUnit unit) throws InterruptedException;
    
        boolean awaitUntil(Date deadline) throws InterruptedException;
    
        void signal();
    
        void signalAll();
    
    }

ConditionObject实现了Condition接口中的等待和通知接口，在这个内部类里面，通过维护两个变量`firstWaiter`和`lastWaiter`来维护一个FIFO队列:

public class ConditionObject implements Condition, java.io.Serializable { 
        /** First node of condition queue. */
        private transient Node firstWaiter;
        /** Last node of condition queue. */
        private transient Node lastWaiter;
    }

> 可以看到等待队列也是复用了AQS的Node内部类来构造节点

### 主要方法说明 ###

<table> 
 <thead> 
  <tr> 
   <th align="left">方法</th> 
   <th align="left">说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">void await()</td> 
   <td align="left">让持有锁的线程进入等待队列并释放锁（从同步队列中移除）</td> 
  </tr> 
  <tr> 
   <td align="left">void awaitUninterruptibly()</td> 
   <td align="left">同await()方法，区别在于该方法不支持中断</td> 
  </tr> 
  <tr> 
   <td align="left">long awaitNanos(long nanosTimeout)</td> 
   <td align="left">实现超时等待，单位是纳秒</td> 
  </tr> 
  <tr> 
   <td align="left">boolean await(long time, TimeUnit unit</td> 
   <td align="left">实现超时等待，可指定时间刻度</td> 
  </tr> 
  <tr> 
   <td align="left">boolean awaitUntil(Date deadline)</td> 
   <td align="left">等待直到某个时间点</td> 
  </tr> 
  <tr> 
   <td align="left">void signal()</td> 
   <td align="left">唤醒处于等待队列中首节点线程，使其加入到同步队列中，等待竞争同步锁</td> 
  </tr> 
  <tr> 
   <td align="left">void signalAll()</td> 
   <td align="left">唤醒处于等待队列中所有节点线程，使其全部加入到同步队列中，等待竞争同步锁</td> 
  </tr> 
 </tbody> 
</table>

### 实现分析 ###

#### await分析 ####

入口方法在ConditionObject的`await`方法:

public final void await() throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        // 节点加入等待队列尾部
        Node node = addConditionWaiter();
        // 释放独占锁同步状态
        int savedState = fullyRelease(node);
        int interruptMode = 0;
        // 当前节点没有被移到同步队列(仍在等待队列中)
        while (!isOnSyncQueue(node)) {
            // 阻塞当前线程
            LockSupport.park(this);
            // 检查中断状态，若中断，则跳出while循环
            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);
    }

主要逻辑说明如下:

1) 首先会调用`addConditionWaiter`方法将节点加入等待队列，然后释放锁  
2) 之后将节点阻塞并循环判断是否节点被唤醒（唤醒后会节点会移动到同步队列，唤醒由signal或signalAll实现）。  
3) 当节点被唤醒，需要再次判断是否能够竞争到锁。

温馨提示: `addConditionWaiter`方法没有用到CAS将节点加入，是因为能够调用`await`方法的线程肯定是持有独占锁的线程，确保只有一个线程执行入等待队列操作，非多线程并发操作，无需使用CAS

`fullyRelease`方法(AQS中的方法)主要是释放同步状态，并将持有的同步状态返回，以便后面被唤醒后，再次竞争同步状态。

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

> 其他await方法不在此赘述，主要逻辑大同小异。

#### signal ####

只负责唤醒一个节点（非CANCEL）

public final void signal() {
        // 必须是独占模式，并且持有锁
        if (!isHeldExclusively())
            throw new IllegalMonitorStateException();
        Node first = firstWaiter;
        if (first != null)
            doSignal(first);
    }

`doSignal`方法会负责找第一个非CANCEL节点，将它从等待队列中移除，并加到同步队列尾部:

private void doSignal(Node first) {
        do {
            if ( (firstWaiter = first.nextWaiter) == null)
                lastWaiter = null;
            first.nextWaiter = null;
        } while (!transferForSignal(first) &&
                 (first = firstWaiter) != null);
    }

`transferForSignal`方法:

final boolean transferForSignal(Node node) {
        // 如果当前节点waitStatus不是CONDITION，返回失败，CAS成功则将waitStatus重置为0，
        // 以便下面将节点加回同步队列
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            return false;
    
        // 节点加回同步队列，并进行当前线程唤醒（前提是前置节点状态非CANCEL???），
        // 以便于当前阻塞在await方法的线程能够从await方法返回，去竞争同步锁
        Node p = enq(node);
        int ws = p.waitStatus;
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
            LockSupport.unpark(node.thread);
        return true;
    }

#### signalAll ####

看方法名其实也能猜个大概，比起`signal`方法只能唤醒一个阻塞线程，对于`signalAll`方法，是能够唤醒所有阻塞在同步队列中的线程，并将其全部移到同步队列中。

public final void signalAll() {
        // 同样需要判断持有独占锁
        if (!isHeldExclusively())
            throw new IllegalMonitorStateException();
        Node first = firstWaiter;
        if (first != null)
            doSignalAll(first);
    }
    
    
    private void doSignalAll(Node first) {
        lastWaiter = firstWaiter = null;
        // 将所有等待队列中非CANCEL的节点移到同步队列中并唤醒
        do {
            Node next = first.nextWaiter;
            first.nextWaiter = null;
            transferForSignal(first);
            first = next;
        } while (first != null);
    }

## 总结 ##

1.  AQS是实现JUC同步组件的抽象基础，实现的组件主要包括独占锁`ReentrantLock`和共享锁`ReentrantReadWriteLock`，以及在共享实现的基础上，实现多种同步组件，如`CountDownLatch`, `Semaphore`等
2.  AQS内部实际可以抽象为一个同步队列和多个条件队列，其中同步队列用于实现不同的同步组件功能；而条件队列用于实现类似Object提供的wait/notify机制

> synchronized + wait/notify 高级版为 JUC的 Lock + Condition await/signal，其实现细节就在AQS里面

[https_www.cnblogs.com_dolphin0520_p_3920373.html]: https://www.cnblogs.com/dolphin0520/p/3920373.html
[AQS_sync_queue]: http://7xtamf.com1.z0.glb.clouddn.com//image/blog/jdk_aqs_introduction/AQS_sync_queue.png
[https_segmentfault.com_a_1190000008420938]: https://segmentfault.com/a/1190000008420938
[AQS_wait_queue]: http://7xtamf.com1.z0.glb.clouddn.com//image/blog/jdk_aqs_introduction/AQS_wait_queue.png