【并发编程】【JDK源码】J.U.C--AQS (AbstractQueuedSynchronizer)(1/2)

J.U.C实现基础

AQS、非阻塞数据结构和原子变量类（java.util.concurrent.atomic包中的类），concurrent包中的基础类都是使用这种模式来实现的。而concurrent包中的高层类又是依赖于这些基础类来实现的。从整体来看，concurrent包的实现示意图如下：

Java中的许多可阻塞类，例如ReentrantLock、Semaphore、ReentrantReadWriteLock、CountDownLatch等，都是基于AQS构建的。

注：在jdk 1.8.0_111源码中FutureTask不是基于AQS的，而是基于CAS

FutureTask源码注释：
    /*
     * Revision notes: This differs from previous versions of this
     * class that relied on AbstractQueuedSynchronizer, mainly to
     * avoid surprising users about retaining interrupt status during
     * cancellation races. Sync control in the current design relies
     * on a "state" field updated via CAS to track completion, along
     * with a simple Treiber stack to hold waiting threads.
     *
     * Style note: As usual, we bypass overhead of using
     * AtomicXFieldUpdaters and instead directly use Unsafe intrinsics.
     */

AQS在源码中的位置

AQS

什么是AQS

AQS(AbstractQueuedSynchronizer)，AQS是JDK下提供的一套用于实现基于FIFO等待队列的阻塞锁和相关的同步器的一个同步框架。这个抽象类被设计为作为一些可用原子int值来表示状态的同步器的基类。如果你有看过类似 CountDownLatch 类的源码实现，会发现其内部有一个继承了 AbstractQueuedSynchronizer 的内部类 Sync 。可见 CountDownLatch 是基于AQS框架来实现的一个同步器.类似的同步器在JUC下还有不少。(eg. Semaphore )

AQS用法

如上所述，AQS管理一个关于状态信息的单一整数，该整数可以表现任何状态。比如， Semaphore 用它来表现剩余的许可数，ReentrantLock用它来表现拥有它的线程已经请求了多少次锁；FutureTask 用它来表现任务的状态(尚未开始、运行、完成和取消)。

JDK源码中描述如下：

Provides a framework for implementing blocking locks and related
synchronizers (semaphores, events, etc) that rely on
first-in-first-out (FIFO) wait queues. This class is designed to
be a useful basis for most kinds of synchronizers that rely on a
single atomic {@code int} value to represent state. Subclasses
must define the protected methods that change this state, and which
define what that state means in terms of this object being acquired
or released. Given these, the other methods in this class carry
out all queuing and blocking mechanics. Subclasses can maintain
other state fields, but only the atomically updated {@code int}
value manipulated using methods {@link #getState}, {@link
#setState} and {@link #compareAndSetState} is tracked with respect
to synchronization.

Subclasses should be defined as non-public internal helper
classes that are used to implement the synchronization properties
of their enclosing class. Class
{@code AbstractQueuedSynchronizer} does not implement any
synchronization interface. Instead it defines methods such as
{@link #acquireInterruptibly} that can be invoked as
appropriate by concrete locks and related synchronizers to
implement their public methods.

This class supports either or both a default exclusive
mode and a shared mode. When acquired in exclusive mode,
attempted acquires by other threads cannot succeed. Shared mode
acquires by multiple threads may (but need not) succeed. This class
does not "understand" these differences except in the
mechanical sense that when a shared mode acquire succeeds, the next
waiting thread (if one exists) must also determine whether it can
acquire as well. Threads waiting in the different modes share the
same FIFO queue. Usually, implementation subclasses support only
one of these modes, but both can come into play for example in a
{@link ReadWriteLock}. Subclasses that support only exclusive or
only shared modes need not define the methods supporting the unused mode.

This class defines a nested {@link ConditionObject} class that
can be used as a {@link Condition} implementation by subclasses
supporting exclusive mode for which method {@link
#isHeldExclusively} reports whether synchronization is exclusively
held with respect to the current thread, method {@link #release}
invoked with the current {@link #getState} value fully releases
this object, and {@link #acquire}, given this saved state value,
eventually restores this object to its previous acquired state. No
{@code AbstractQueuedSynchronizer} method otherwise creates such a
condition, so if this constraint cannot be met, do not use it. The
behavior of {@link ConditionObject} depends of course on the
semantics of its synchronizer implementation.

This class provides inspection, instrumentation, and monitoring
methods for the internal queue, as well as similar methods for
condition objects. These can be exported as desired into classes
using an {@code AbstractQueuedSynchronizer} for their
synchronization mechanics.

Serialization of this class stores only the underlying atomic
integer maintaining state, so deserialized objects have empty
thread queues. Typical subclasses requiring serializability will
define a {@code readObject} method that restores this to a known
initial state upon deserialization.

Usage

To use this class as the basis of a synchronizer, redefine the
following methods, as applicable, by inspecting and/or modifying
the synchronization state using {@link #getState}, {@link
#setState} and/or {@link #compareAndSetState}:

{@link #tryAcquire}
{@link #tryRelease}
{@link #tryAcquireShared}
{@link #tryReleaseShared}
{@link #isHeldExclusively}
Each of these methods by default throws {@link
UnsupportedOperationException}. Implementations of these methods
must be internally thread-safe, and should in general be short and
not block. Defining these methods is the only supported
means of using this class. All other methods are declared
{@code final} because they cannot be independently varied.

You may also find the inherited methods from {@link
AbstractOwnableSynchronizer} useful to keep track of the thread
owning an exclusive synchronizer. You are encouraged to use them
-- this enables monitoring and diagnostic tools to assist users in
determining which threads hold locks.

Even though this class is based on an internal FIFO queue, it
does not automatically enforce FIFO acquisition policies. The core
of exclusive synchronization takes the form:

Acquire:

while (!tryAcquire(arg)) {

enqueue thread if it is not already queued;

possibly block current thread;

}

Release:

if (tryRelease(arg))

unblock the first queued thread;

(Shared mode is similar but may involve cascading signals.)

Because checks in acquire are invoked before
enqueuing, a newly acquiring thread may barge ahead of
others that are blocked and queued. However, you can, if desired,
define {@code tryAcquire} and/or {@code tryAcquireShared} to
disable barging by internally invoking one or more of the inspection
methods, thereby providing a fair FIFO acquisition order.
In particular, most fair synchronizers can define {@code tryAcquire}
to return {@code false} if {@link #hasQueuedPredecessors} (a method
specifically designed to be used by fair synchronizers) returns
{@code true}. Other variations are possible.

Throughput and scalability are generally highest for the
default barging (also known as greedy,
renouncement, and convoy-avoidance) strategy.
While this is not guaranteed to be fair or starvation-free, earlier
queued threads are allowed to recontend before later queued
threads, and each recontention has an unbiased chance to succeed
against incoming threads. Also, while acquires do not
"spin" in the usual sense, they may perform multiple
invocations of {@code tryAcquire} interspersed with other
computations before blocking. This gives most of the benefits of
spins when exclusive synchronization is only briefly held, without
most of the liabilities when it is not. If so desired, you can
augment this by preceding calls to acquire methods with
"fast-path" checks, possibly prechecking {@link #hasContended}
and/or {@link #hasQueuedThreads} to only do so if the synchronizer
is likely not to be contended.

This class provides an efficient and scalable basis for
synchronization in part by specializing its range of use to
synchronizers that can rely on {@code int} state, acquire, and
release parameters, and an internal FIFO wait queue. When this does
not suffice, you can build synchronizers from a lower level using
{@link java.util.concurrent.atomic atomic} classes, your own custom
{@link java.util.Queue} classes, and {@link LockSupport} blocking
support.

Usage Examples

Here is a non-reentrant mutual exclusion lock class that uses
the value zero to represent the unlocked state, and one to
represent the locked state. While a non-reentrant lock
does not strictly require recording of the current owner
thread, this class does so anyway to make usage easier to monitor.
It also supports conditions and exposes
one of the instrumentation methods:

 {@code

class Mutex implements Lock, java.io.Serializable {

// Our internal helper class
private static class Sync extends AbstractQueuedSynchronizer {
  // Reports whether in locked state
  protected boolean isHeldExclusively() {
    return getState() == 1;
  }

  // Acquires the lock if state is zero
  public boolean tryAcquire(int acquires) {
    assert acquires == 1; // Otherwise unused
    if (compareAndSetState(0, 1)) {
      setExclusiveOwnerThread(Thread.currentThread());
      return true;
    }
    return false;
  }

  // Releases the lock by setting state to zero
  protected boolean tryRelease(int releases) {
    assert releases == 1; // Otherwise unused
    if (getState() == 0) throw new IllegalMonitorStateException();
    setExclusiveOwnerThread(null);
    setState(0);
    return true;
  }

  // Provides a Condition
  Condition newCondition() { return new ConditionObject(); }

  // Deserializes properly
  private void readObject(ObjectInputStream s)
      throws IOException, ClassNotFoundException {
    s.defaultReadObject();
    setState(0); // reset to unlocked state
  }
}

// The sync object does all the hard work. We just forward to it.
private final Sync sync = new Sync();

public void lock()                { sync.acquire(1); }
public boolean tryLock()          { return sync.tryAcquire(1); }
public void unlock()              { sync.release(1); }
public Condition newCondition()   { return sync.newCondition(); }
public boolean isLocked()         { return sync.isHeldExclusively(); }
public boolean hasQueuedThreads() { return sync.hasQueuedThreads(); }
public void lockInterruptibly() throws InterruptedException {
  sync.acquireInterruptibly(1);
}
public boolean tryLock(long timeout, TimeUnit unit)
    throws InterruptedException {
  return sync.tryAcquireNanos(1, unit.toNanos(timeout));
}

}}

Here is a latch class that is like a
{@link java.util.concurrent.CountDownLatch CountDownLatch}
except that it only requires a single {@code signal} to
fire. Because a latch is non-exclusive, it uses the {@code shared}
acquire and release methods.

 {@code

class BooleanLatch {

private static class Sync extends AbstractQueuedSynchronizer {
  boolean isSignalled() { return getState() != 0; }

  protected int tryAcquireShared(int ignore) {
    return isSignalled() ? 1 : -1;
  }

  protected boolean tryReleaseShared(int ignore) {
    setState(1);
    return true;
  }
}

private final Sync sync = new Sync();
public boolean isSignalled() { return sync.isSignalled(); }
public void signal()         { sync.releaseShared(1); }
public void await() throws InterruptedException {
  sync.acquireSharedInterruptibly(1);
}

}}

@since 1.5
@author Doug Lea

如JDK的文档中所说，使用AQS来实现一个同步器需要覆盖实现如下几个方法，并且使用getState,setState,compareAndSetState这几个方法来设置获取状态

boolean tryAcquire(int arg)
boolean tryRelease(int arg)
int tryAcquireShared(int arg)
boolean tryReleaseShared(int arg)
boolean isHeldExclusively()

以上方法不需要全部实现，根据获取的锁的种类可以选择实现不同的方法，支持独占(排他)获取锁的同步器应该实现tryAcquire、tryRelease、isHeldExclusively而支持共享获取的同步器应该实现tryAcquireShared、tryReleaseShared、isHeldExclusively。下面以 CountDownLatch 举例说明基于AQS实现同步器, CountDownLatch 用同步状态持有当前计数，countDown方法调用 release从而导致计数器递减；当计数器为0时，解除所有线程的等待；await调用acquire，如果计数器为0，acquire 会立即返回，否则阻塞。通常用于某任务需要等待其他任务都完成后才能继续执行的情景。源码如下:

public class CountDownLatch {
    /**
     * 基于AQS的内部Sync
     * 使用AQS的state来表示计数count.
     */
    private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;
        Sync(int count) {
            // 使用AQS的getState()方法设置状态
            setState(count);
        }
        int getCount() {
            // 使用AQS的getState()方法获取状态
            return getState();
        }
        // 覆盖在共享模式下尝试获取锁
        protected int tryAcquireShared(int acquires) {
            // 这里用状态state是否为0来表示是否成功，为0的时候可以获取到返回1，否则不可以返回-1
            return (getState() == 0) ? 1 : -1;
        }
        // 覆盖在共享模式下尝试释放锁
        protected boolean tryReleaseShared(int releases) {
            // 在for循环中Decrement count直至成功;
            // 当状态值即count为0的时候，返回false表示 signal when transition to zero
            for (;;) {
                int c = getState();
                if (c == 0)
                    return false;
                int nextc = c-1;
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }
    }
    private final Sync sync;
    // 使用给定计数值构造CountDownLatch
    public CountDownLatch(int count) {
        if (count < 0) throw new IllegalArgumentException("count < 0");
        this.sync = new Sync(count);
    }
    // 让当前线程阻塞直到计数count变为0，或者线程被中断
    public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }
    // 阻塞当前线程，除非count变为0或者等待了timeout的时间。当count变为0时，返回true
    public boolean await(long timeout, TimeUnit unit)
        throws InterruptedException {
        return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
    }
    // count递减
    public void countDown() {
        sync.releaseShared(1);
    }
    // 获取当前count值
    public long getCount() {
        return sync.getCount();
    }
    public String toString() {
        return super.toString() + "[Count = " + sync.getCount() + "]";
    }
}

Doug Lea曾经介绍过 AQS 的设计初衷。从原理上，一种同步结构往往是可以利用其他的结构实现的，例如可以使用 Semaphore实现互斥锁。但是，对某种同步结构的倾向，会导致复杂、晦涩的实现逻辑，所以，他选择了将基础的同步相关操作抽象在 AbstractQueuedSynchronizer 中，利用 AQS 为我们构建同步结构提供了范本。

AQS 内部数据和方法，可以简单拆分为：

• 一个 volatile 的整数成员表征状态，同时提供了 setState 和 getState 方法

private volatile int state;

• 一个先入先出（FIFO）的等待线程队列，以实现多线程间竞争和等待，这是 AQS 机制的核心之一。

• 各种基于 CAS 的基础操作方法，以及各种期望具体同步结构去实现的 acquire/release 方法。

利用 AQS 实现一个同步结构，至少要实现两个基本类型的方法，分别是 acquire操作，获取资源的独占权；还有就是 release 操作，释放对某个资源的独占。

以 ReentrantLock 为例，它内部通过扩展 AQS 实现了 Sync 类型，以 AQS 的 state来反映锁的持有情况。

private final Sync sync;
abstract static class Sync extends AbstractQueuedSynchronizer { …}

下面是 ReentrantLock 对应 acquire 和 release 操作，如果是 CountDownLatch 则可以看作是 await()/countDown()，具体实现也有区别。

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

排除掉一些细节，整体地分析 acquire 方法逻辑，其直接实现是在 AQS 内部，调用了 tryAcquire 和 acquireQueued，这是两个需要搞清楚的基本部分。

public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

首先，我们来看看 tryAcquire。在 ReentrantLock 中，tryAcquire 逻辑实现在 NonfairSync 和 FairSync 中，分别提供了进一步的非公平或公平性方法，而 AQS 内部 tryAcquire 仅仅是个接近未实现的方法（直接抛异常），这是留个实现者自己定义的操作。

我们可以看到公平性在 ReentrantLock 构建时如何指定的，具体如下：

public ReentrantLock() {
        sync = new NonfairSync(); // 默认是非公平的
    }
    public ReentrantLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
    }

以非公平的 tryAcquire 为例，其内部实现了如何配合状态与 CAS获取锁，注意，对比公平版本的tryAcquire，它在锁无人占有时，并不检查是否有其他等待者，这里体现了非公平的语义。

final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();// 获取当前 AQS 内部状态量
    if (c == 0) { // 0 表示无人占有，则直接用 CAS 修改状态位，
        if (compareAndSetState(0, acquires)) {// 不检查排队情况，直接争抢
            setExclusiveOwnerThread(current);  // 并设置当前线程独占锁
            return true;
        }
    } else if (current == getExclusiveOwnerThread()) { // 即使状态不是 0，也可能当前线程是锁持有者，因为这是再入锁
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

接下来再来分析 acquireQueued，如果前面的 tryAcquire失败，代表着锁争抢失败，进入排队竞争阶段。这里就是我们所说的，利用 FIFO 队列，实现线程间对锁的竞争的部分，算是是 AQS 的核心逻辑。

当前线程会被包装成为一个排他模式的节点（EXCLUSIVE），通过 addWaiter 方法添加到队列中。acquireQueued的逻辑，简要来说，就是如果当前节点的前面是头节点，则试图获取锁，一切顺利则成为新的头节点；否则，有必要则等待，具体处理逻辑请参考我添加的注释。

final boolean acquireQueued(final Node node, int arg) {
      boolean interrupted = false;
      try {
        for (;;) {// 循环
            final Node p = node.predecessor();// 获取前一个节点
            if (p == head && tryAcquire(arg)) { // 如果前一个节点是头结点，表示当前节点合适去 tryAcquire
                setHead(node); // acquire 成功，则设置新的头节点
                p.next = null; // 将前面节点对当前节点的引用清空
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node)) // 检查是否失败后需要 park
                interrupted |= parkAndCheckInterrupt();
        }
       } catch (Throwable t) {
        cancelAcquire(node);// 出现异常，取消
        if (interrupted)
                selfInterrupt();
        throw t;
      }
}

到这里线程试图获取锁的过程基本展现出来了，tryAcquire是按照特定场景需要开发者去实现的部分，而线程间竞争则是 AQS 通过 Waiter 队列与 acquireQueued 提供的，在 release方法中，同样会对队列进行对应操作。

参考资料：

JAVA并发编程: CAS和AQS

《JAVA并发编程实战》

极客时间：第22讲 | AtomicInteger底层实现原理是什么？如何在自己的产品代码中应用CAS操作

源码剖析AQS在几个同步工具类中的使用