AQS全称是AbstractQueuedSynchronizer，在Java中大部分所使用到的阻塞锁/同步器都是基于它实现的。
介绍
AQS，即AbstractQueuedSynchronizer，JDK官方文档是这样描述AQS的，AQS是基于FIFO（first-in-first-out）队列来实现阻塞锁和同步器的框架。例如，我们日常开发中使用到的ReentrantLock就是基于AQS来实现的，而从其源码也可以看到ReentrantLock主要的实现逻辑也是放到AQS中的。
我们不妨先来看看官方文档是如何对AQS进行描述的：
/**
* 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.
*
* ...
*
* <p>This class supports either or both a default <em>exclusive</em>
* mode and a <em>shared</em> 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.
*
* <p>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.
*
* <p>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.
*
* ...
*
* <h3>Usage</h3>
*
* <p>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}:
*
* <ul>
* <li> {@link #tryAcquire}
* <li> {@link #tryRelease}
* <li> {@link #tryAcquireShared}
* <li> {@link #tryReleaseShared}
* <li> {@link #isHeldExclusively}
* </ul>
*
* 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 <em>only</em> supported
* means of using this class. All other methods are declared
* {@code final} because they cannot be independently varied.
*
* <p>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.
*
* ...
*
*/
public abstract class AbstractQueuedSynchronizer
extends AbstractOwnableSynchronizer
implements java.io.Serializable {
}
为了便于理解，笔者对AQS相关的注释描述进行了整理，并将其分为两个部分：
第一部分是关于AQS的设计
AQS的同步机制依赖于原子变量state。AQS的阻塞机制依赖于FIFO等待队列。AQS支持排他模式(EXCLUSIVE)（默认）和共享模式(SHARED)。AQS支持条件队列Condition，并定义了Condition的实现类CondtionObject。
所谓排他模式(
EXCLUSIVE)表示的是以独占的方式获取资源，即当资源被持有的情况下，其他线程是无法（成功）获取的；而共享模式(SHARED)则表示以共享的方式获取资源，即当资源被持有的情况下，其他线程也是可以获取的（但不一定能成功）。
第二部分是关于AQS的实现
在AQS的实现类中，我们需要定义如下protected级别的方法，分别是tryAcquire、tryRelease、tryAcquireShared、tryReleaseShared和isHeldExclusively（可通过检查/修改state实现不同的特性），默认情况下这些方法都会抛出UnsupportedOperationException异常。此处需要注意，在这些方法的实现中需要保证其线程安全、短暂和非阻塞。
对于上述的
protected方法没有使用abstract修饰是为了让开发者尽可能少的开发代码，因为并不是所有阻塞锁/同步器都需要实现所有的protected方法，而是会根据排他模式（EXCLUSIVE）或者共享模式（SHARED）来选择需要实现的方法。
另外，在AQS中会通过继承AbstractOwnableSynchronizer来实现资源持有线程的轨迹记录（可用于实现可重入机制）。
/**
* A synchronizer that may be exclusively owned by a thread. This
* class provides a basis for creating locks and related synchronizers
* that may entail a notion of ownership. The
* {@code AbstractOwnableSynchronizer} class itself does not manage or
* use this information. However, subclasses and tools may use
* appropriately maintained values to help control and monitor access
* and provide diagnostics.
*
* ...
*/
public abstract class AbstractOwnableSynchronizer
implements java.io.Serializable {
/**
* Empty constructor for use by subclasses.
*/
protected AbstractOwnableSynchronizer() {}
/**
* The current owner of exclusive mode synchronization.
*/
private transient Thread exclusiveOwnerThread;
/**
* Sets the thread that currently owns exclusive access.
* A {@code null} argument indicates that no thread owns access.
* This method does not otherwise impose any synchronization or
* {@code volatile} field accesses.
* @param thread the owner thread
*/
protected final void setExclusiveOwnerThread(Thread thread) {
exclusiveOwnerThread = thread;
}
/**
* Returns the thread last set by {@code setExclusiveOwnerThread},
* or {@code null} if never set. This method does not otherwise
* impose any synchronization or {@code volatile} field accesses.
* @return the owner thread
*/
protected final Thread getExclusiveOwnerThread() {
return exclusiveOwnerThread;
}
}
AbstractOwnableSynchronizer的代码相对比较简短，阅读起来相对容易理解。从代码层面上看就是提供一个地方让我们可以把线程暂存储起来。另外，通过这种方式我们可以实现可重入机制（使用exclusiveOwnerThread变量记录当前持有线程，并将其与后续获取线程进行比较），或者保证资源获取与释放线程的一致性等。
至此，通过官方文档的描述我们大概能了解AQS中的一些实现机制，但是并没有得出十分清晰的原理性结论，因此下文笔者将进一步对AQS原理进行剖析。
同步语义
在AQS中，我们可以通过原子变量state来实现其同步语义，即借助state变量来实现对资源获取与释放的原子性，进而实现最终的同步语义。
从资源的角度上看，我们完全可以将
state变量比作是可获取的资源，而对state变量的修改则可等价为对资源的获取与释放。
其中，对于state变量的原子性则是通过volatile关键字和CAS机制的组合来实现的，所以对它的获取、设置和更新需要用到特定的方法，即getState、setState和compareAndSetState。
public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {
private volatile int state;
protected final int getState() {
return state;
}
protected final void setState(int newState) {
state = newState;
}
protected final boolean compareAndSetState(int expect, int update) {
// See below for intrinsics setup to support this
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}
}
但是，AQS并没有直接通过这种方式对资源的获取/释放进行硬编码（强制绑定关系与逻辑），而是提供了一系列抽象的protected方法让我们可以自行对资源获取与释放进行定义（自由使用state变量）。下面笔者总结了这几个方法各自的作用：
| 方法 | 描述 |
|---|---|
tryAcquire |
表示在排他模式(EXCLUSIVE)下去获取资源，如果返回true表示获取成功，否则表示获取失败。其中，在方法的实现中我们应该判断当前是否能在独占模式获取资源。 |
tryRelease |
表示在排他模式(EXCLUSIVE)下去释放资源，如果返回true表示全部释放成功，否则表示释放失败或者部分释放。 |
tryAcquireShared |
表示在共享模式(SHARED)下去去获取资源，如果返回大于0表示获取成功并且其后继节点也可能成功获取资源；如果返回等于0表示获取成功但其后继节点不能再成功获取资源了；如果返回小于0则表示获取失败。其中，在方法的实现中我们应该判断当前是否能够在共享模式下获取资源。 |
tryReleaseShared |
表示在共享模式(SHARED)下去释放资源，如果返回true表示释放成功，否则表示释放失败。 |
isHeldExclusively |
表示资源是否被独占地持有，如果返回true表示被独占持有，否则表示没有被独占持有。 |
综上所述，当要实现独占语义时需要实现tryAcquire、tryRelease和isHeldExclusively；当要实现共享语义时则需要实现tryAcquireShared和tryReleaseShared。而对这些方法的实现就是我们使用AQS框架所需要做的所有工作了。
通过上述抽象定义，我们就可以自由地实现一次只能持有一个资源的同步器/锁（
ReentrantLock），或者同时可以持有多个资源的同步器/锁（Semaphore）。
在对AQS的同步语义有一定的了解后，我们再来看看它是如何通过FIFO队列来实现阻塞机制的。
数据结构
对于AQS的阻塞机制，它是采用FIFO等待队列来实现的，在线程获取资源失败后，将它插入到FIFO等待队列中进行等待，直到条件符合时再次唤醒队列中等待的线程来获取资源。下面我们来看看它的数据结构Node：
在
AQS的条件等待机制中也采用了FIFO队列来实现，其中数据结构同样使用了Node。
/**
* Wait queue node class.
*
* <p>The wait queue is a variant of a "CLH" (Craig, Landin, and
* Hagersten) lock queue. CLH locks are normally used for
* spinlocks. We instead use them for blocking synchronizers, but
* use the same basic tactic of holding some of the control
* information about a thread in the predecessor of its node. A
* "status" field in each node keeps track of whether a thread
* should block. A node is signalled when its predecessor
* releases. Each node of the queue otherwise serves as a
* specific-notification-style monitor holding a single waiting
* thread. The status field does NOT control whether threads are
* granted locks etc though. A thread may try to acquire if it is
* first in the queue. But being first does not guarantee success;
* it only gives the right to contend. So the currently released
* contender thread may need to rewait.
*
* <p>To enqueue into a CLH lock, you atomically splice it in as new
* tail. To dequeue, you just set the head field.
* <pre>
* +------+ prev +-----+ +-----+
* head | | <---- | | <---- | | tail
* +------+ +-----+ +-----+
* </pre>
*
* <p>Insertion into a CLH queue requires only a single atomic
* operation on "tail", so there is a simple atomic point of
* demarcation from unqueued to queued. Similarly, dequeuing
* involves only updating the "head". However, it takes a bit
* more work for nodes to determine who their successors are,
* in part to deal with possible cancellation due to timeouts
* and interrupts.
*
* <p>The "prev" links (not used in original CLH locks), are mainly
* needed to handle cancellation. If a node is cancelled, its
* successor is (normally) relinked to a non-cancelled
* predecessor. For explanation of similar mechanics in the case
* of spin locks, see the papers by Scott and Scherer at
* http://www.cs.rochester.edu/u/scott/synchronization/
*
* <p>We also use "next" links to implement blocking mechanics.
* The thread id for each node is kept in its own node, so a
* predecessor signals the next node to wake up by traversing
* next link to determine which thread it is. Determination of
* successor must avoid races with newly queued nodes to set
* the "next" fields of their predecessors. This is solved
* when necessary by checking backwards from the atomically
* updated "tail" when a node's successor appears to be null.
* (Or, said differently, the next-links are an optimization
* so that we don't usually need a backward scan.)
*
* <p>Cancellation introduces some conservatism to the basic
* algorithms. Since we must poll for cancellation of other
* nodes, we can miss noticing whether a cancelled node is
* ahead or behind us. This is dealt with by always unparking
* successors upon cancellation, allowing them to stabilize on
* a new predecessor, unless we can identify an uncancelled
* predecessor who will carry this responsibility.
*
* <p>CLH queues need a dummy header node to get started. But
* we don't create them on construction, because it would be wasted
* effort if there is never contention. Instead, the node
* is constructed and head and tail pointers are set upon first
* contention.
*
* <p>Threads waiting on Conditions use the same nodes, but
* use an additional link. Conditions only need to link nodes
* in simple (non-concurrent) linked queues because they are
* only accessed when exclusively held. Upon await, a node is
* inserted into a condition queue. Upon signal, the node is
* transferred to the main queue. A special value of status
* field is used to mark which queue a node is on.
*
* ...
*
*/
static final class Node {
// ...
}
通过上述的注释说明应该可以清晰地认识到其设计思想，这里笔者对这段注释做了一个简单的翻译：
AQS的等待队列（wait queue）是CLH锁（自旋锁）队列的一个变种。队列中每个节点Node都会存储一个线程变量来标识当前执行线程和一个状态waitStatus来标记当前线程是否被阻塞。一个节点会在其前驱节点（一般为head节点）释放后再次被唤醒，但是它并不一定能成功获取资源（只是提供了竞争的机会，如果失败会再次进入等待状态），这取决于实现的语义是否公平。
CLH锁队列，可以参考论文《A Hierarchical CLH Queue Lock》
在队列中，节点出队操作只需简单地设置head节点，但节点入队操作则需要对tail节点进行原子性操作。而对于节点之间的连接则是通过prev链接和next链接实现，其中prev链接主要用于处理节点的取消操作，即当节点被取消时其后继节点需重新链接到前面离它最近的非取消节点；而next链接则是用于实现阻塞机制，即当节点被释放后通过next链接来唤醒后继节点（每个节点中都存储着阻塞的线程）。另外，在唤醒后继节点的操作和新节点入队的操作发生竞争时，如果只是通过next链接往后遍历可能会出现null，这时需要从tail节点向前遍历（因为入队时会先设置prev链接）。
关于
head节点的创建，采用的是一种延迟加载的方式，在第一个（阻塞）节点入队时才进行初始化，避免在不存在阻塞等待的情况下造成不必要的浪费。
另外，通过条件等待机制所使用到的在条件队列(Condition)也是采用了相同的数据结构Node，其中它们之间的区别在于条件队列并没有使用next和prev链接，而是使用了额外的nextWaiter链接。在实现上，因为条件队列需要在独占持有资源的情况下才能访问，所以只需简单地将节点链接到条件队列即可（无并发）。在使用上，await方法会让线程进入条件等待状态，即将节点从等待队列转移到条件队列；signal方法则是会将线程从条件等待状态中唤醒，即将节点从条件队列转移到等待队列。
对于节点
Node存储在哪种队列（等待队列或条件队列）则是通过waitStatus字段来标记。
通过官方的注释，我们应该对等待/条件队列及其数据结构Node具有一定的认识了，接下来我们再从源码的角度分析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;
/**
* Status field, taking on only the values:
* SIGNAL: The successor of this node is (or will soon be)
* blocked (via park), so the current node must
* unpark its successor when it releases or
* cancels. To avoid races, acquire methods must
* first indicate they need a signal,
* then retry the atomic acquire, and then,
* on failure, block.
* CANCELLED: This node is cancelled due to timeout or interrupt.
* Nodes never leave this state. In particular,
* a thread with cancelled node never again blocks.
* CONDITION: This node is currently on a condition queue.
* It will not be used as a sync queue node
* until transferred, at which time the status
* will be set to 0. (Use of this value here has
* nothing to do with the other uses of the
* field, but simplifies mechanics.)
* PROPAGATE: A releaseShared should be propagated to other
* nodes. This is set (for head node only) in
* doReleaseShared to ensure propagation
* continues, even if other operations have
* since intervened.
* 0: None of the above
*
* The values are arranged numerically to simplify use.
* Non-negative values mean that a node doesn't need to
* signal. So, most code doesn't need to check for particular
* values, just for sign.
*
* The field is initialized to 0 for normal sync nodes, and
* CONDITION for condition nodes. It is modified using CAS
* (or when possible, unconditional volatile writes).
*/
volatile int waitStatus;
/**
* Link to predecessor node that current node/thread relies on
* for checking waitStatus. Assigned during enqueuing, and nulled
* out (for sake of GC) only upon dequeuing. Also, upon
* cancellation of a predecessor, we short-circuit while
* finding a non-cancelled one, which will always exist
* because the head node is never cancelled: A node becomes
* head only as a result of successful acquire. A
* cancelled thread never succeeds in acquiring, and a thread only
* cancels itself, not any other node.
*/
volatile Node prev;
/**
* Link to the successor node that the current node/thread
* unparks upon release. Assigned during enqueuing, adjusted
* when bypassing cancelled predecessors, and nulled out (for
* sake of GC) when dequeued. The enq operation does not
* assign next field of a predecessor until after attachment,
* so seeing a null next field does not necessarily mean that
* node is at end of queue. However, if a next field appears
* to be null, we can scan prev's from the tail to
* double-check. The next field of cancelled nodes is set to
* point to the node itself instead of null, to make life
* easier for isOnSyncQueue.
*/
volatile Node next;
/**
* The thread that enqueued this node. Initialized on
* construction and nulled out after use.
*/
volatile Thread thread;
/**
* Link to next node waiting on condition, or the special
* value SHARED. Because condition queues are accessed only
* when holding in exclusive mode, we just need a simple
* linked queue to hold nodes while they are waiting on
* conditions. They are then transferred to the queue to
* re-acquire. And because conditions can only be exclusive,
* we save a field by using special value to indicate shared
* mode.
*/
Node nextWaiter;
/**
* Returns true if node is waiting in shared mode.
*/
final boolean isShared() {
return nextWaiter == SHARED;
}
/**
* Returns previous node, or throws NullPointerException if null.
* Use when predecessor cannot be null. The null check could
* be elided, but is present to help the VM.
*
* @return the predecessor of this node
*/
final Node predecessor() throws NullPointerException {
Node p = prev;
if (p == null)
throw new NullPointerException();
else
return p;
}
Node() { // Used to establish initial head or SHARED marker
}
Node(Thread thread, Node mode) { // Used by addWaiter
this.nextWaiter = mode;
this.thread = thread;
}
Node(Thread thread, int waitStatus) { // Used by Condition
this.waitStatus = waitStatus;
this.thread = thread;
}
}
首先我们先来看waitStatus状态字段在不同取值所表示的意义：
waitStatus |
描述 |
|---|---|
SIGNAL=-1 |
表示当前节点的后继节点需要被唤醒。如果节点被设置为SIGNAL时，那么其被释放或被取消时都会去唤醒后继节点。 |
CANCELLED=1 |
表示当前节点被取消了。在节点被中断或者超时都会设置为CANCELLED。CANCELLED是节点的终态（之一），所以节点进入CANCELLED状态就不能再改变了。 |
CONIDTION=-2 |
表示当前节点处于condition条件队列中等待。当节点从condition条件队列转移到等待队列中时，waitStatus状态将会被设置为0。 |
PROPAGATE=-3 |
表示当前节点在下一次调用acquireShared时应该被无条件的传播。releaseShared也应该被传播到其他节点。 |
INIT=0 |
表示节点初始化状态，只有两个情况waitStatus才会被设置为0，即队尾节点和出队的队头节点。 |
这里的
waitStatus状态是通过数字表示的，这样做是为了简化对代码逻辑的一些处理，比如在唤醒节点时就可以直接通过范围直接进行判断，而不用列举相应的枚举逐个进行判断。
接着是等待队列中用到的两个链接的prev和next：
prev用于链接节点的前驱节点，在节点入队时会被设置，在节点出队时会被设置为null(for GC)。另外，当节点的前驱节点被取消时会重新链接到一个非取消的节点（这个节点总是存在的，因为head节点是不能被取消的）。next用于链接节点的后继节点，在节点入队时会被设置，在节点出队时会被设置为null(for GC)。另外，当节点入队时并不会立刻设置next链接，而是在绑定成功（通过prev链接）之后才设置next链接，所以当发现next指向null时并不代表当前节点就是队尾，对于这种情况AQS会做一个double-check，当发现next指向null时它就会从队尾tail向前遍历。
对于被取消节点的
next链接是指向的是它自己而不是null。
再接着是节点中存储的执行线程thread：
thread属性所表示的是获取资源的线程，它会在节点入队（构造）时被设置，节点出队时被清空（设置为null）。
在节点中保存
thread主要是为了在后续操作中可以很方便地获取该线程并执行相应的操作。
然后是条件队列(Condition)节点间的链接属性nextWaiter：
在条件队列(Condition)中，节点间是通过nextWaiter属性进行链接的。因为条件队列(Condition)只有在独占持有资源时才能被访问，所以进入条件等待时只需要简单地将节点链接到条件队列(Condition)即可。
另外，nextWaiter还可以指向两个特殊值，分别是表示共享模式的SHARED和排他模式的EXCLUSIVE（默认，值为null）。因此在判断是否共享模式的isShared方法中仅仅是通过nextWaiter属性与SHARED进行比较得到最终结果。
最后，这里结合注释和代码内容勾画出等待队列和条件队列的完整结构：
wait队列：
+-------------------+
| +<-----------------------------+
+--------->+ SHARED/EXCLUSIVE | |
| | +<---------+ |
| +-------------------+ | |
| ^ | |
Node | Node | Node | Node |
+------------+ | +------------+ | +------------+ | +------------+ |
| thread | | | thread | | | thread | | | thread | |
| waitStatus | | | waitStatus | | | waitStatus | | | waitStatus | |
| nextWaiter+----+ | nextWaiter+----+ | nextWaiter+----+ | nextWaiter+----+
| | | | | | | |
null<-----+prev +<-------+prev +<-------+prev +<------+prev |
| | | | | | | |
| next+------->+ next+------->+ next+------>+ next+----->null
+------------+ +------------+ +------------+ +------------+
^ ^
| |
head+--------------+ +---------------+tail
condition队列：
Node Node Node Node
+---------------+ +---------------+ +---------------+ +---------------+
| thread | | thread | | thread | | thread |
| waitStatus | | waitStatus | | waitStatus | | waitStatus |
| prev | | prev | | prev | | prev |
| next | | next | | next | | next |
| | | | | | | |
| | | | | | | |
| nextWaiter+--------->+ nextWaiter+---------->+ nextWaiter+--------->+ nextWaiter+--------->null
+-------+-------+ +---------------+ +---------------+ +--------+------+
^ ^
| |
firstWaiter+------+ +-------------+lastWaiter
等待队列
AQS的等待队列是基于CLH锁实现的，CLH锁采用了是一种FIFO+忙自旋的机制来减少资源（锁）的竞争，即在资源竞争失败后进入等待队列，在等待期间不断轮询是否能成功获取资源，如果成功获取则将当前等待节点设置为head节点（即让原head节点出队）。然而因为CLH锁存在自旋消耗CPU的问题，AQS在CLH锁的基础上添加了条件等待的机制（优化），在条件不符合时让队列中的节点（线程）进入线程等待状态，直至head节点完成相应逻辑后唤醒其后继节点（避免了对CPU无效的消耗）。
在对CLH锁及其相关数据结构有一定的认识后，我们再来看看AQS的核心处理流程（为了便于理解，此处修改/删减了部分代码）：
/**
* 排他模式
*/
private void doAcquire(int arg) {
// 入队操作
final Node node = addWaiter(Node.EXCLUSIVE);
try {
// 入队后马上对此节点进行出队操作，并在之后不断进行轮询
for (;;) {
// 此处为队头出队，如果失败就让线程继续等待，否则成功获取资源执行结束（排他模式出队）
final Node p = node.predecessor();
// 此处判断了前驱节点为头节点，所以这里必然只能是公平策略
if (p == head) {
boolean r = tryAcquire(arg);
// 获取资源成功
if(r){
// 将当前节点设置为head节点，并将原head节点出队
setHead(node);
p.next = null; // help GC
return;
}
}
// 判断是否进入等待状态（获取资源失败时）
if (shouldParkAfterFailedAcquire(p, node) && ...){
// 暂停节点线程（由于这里是在一个死循环体内，所以在唤醒之后会继续获取资源，直到成功为止。）
parkThread();
}
}
} finally {
// 如果失败则取消节点
if (failed)
cancelAcquire(node);
}
}
/**
* 共享模式
*/
private void doAcquireShared(int arg) {
// 入队操作
final Node node = addWaiter(Node.SHARED);
try {
// 入队后马上对此节点进行出队操作，并在之后不断进行轮询
for (;;) {
// 此处为队头出队，如果失败就让线程继续等待，否则成功获取资源执行结束（共享模式出队）
final Node p = node.predecessor();
// 此处判断了前驱节点为头节点，所以这里必然只能是公平策略
if (p == head) {
int r = tryAcquireShared(arg);
// 获取资源成功
if (r >= 0) {
// 将当前节点设置为head节点，并将原head节点出队，最后向后传播唤醒信号。
setHeadAndPropagate(node, r);
p.next = null; // help GC
return;
}
}
// 判断是否进入等待状态（获取资源失败时）
if (shouldParkAfterFailedAcquire(p, node) && ...){
// 暂停节点线程（由于这里是在一个死循环体内，所以在唤醒之后会继续获取资源，直到成功为止。）
parkThread();
}
}
} finally {
// 如果失败则取消节点
if (failed)
cancelAcquire(node);
}
}
其中doAcquire方法代表了排他模式下的出队和入队操作，而doAcquireShared方法则代表了共享模式下出队和入队操作。虽然一个是作用于排他模式，另一个是作用于共享模式，但是在宏观上看两者本质上并没有明显的差异，基本上就是在节点在进入等待队列后不断的自旋判断是否能获取到资源，并在资源获取成功后将自己设置为head节点（让原head节点出队）。最后，笔者基于上述代码描绘出其流程的模型图：
+------+ spin / park +------+ spin / park +------+
head +---> | Node | <----------+ | Node | <----------+ | Node | <----+ tail
+---+--+ +---+--+ +------+
| ^
| |
+---------------------+
unpark
至此，我们应该对等待队列的机制和模型有了大体的认识，下面笔者将分别对等待队列的入队操作和出队操作展开进行分析。
节点入队
当执行线程在获取资源失败后，它会被构造成Node节点并链接到等待队列末尾，并进入线程等待状态。而在AQS中，无论是排他模式还是共享模式都是通过addWaiter方法来将Node节点链接到等待队列末尾。下面笔者将addWaiter的相关代码贴了出来：
/**
* Creates and enqueues node for current thread and given mode.
*
* @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared
* @return the new node
*/
private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
Node pred = tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
/**
* Inserts node into queue, initializing if necessary. See picture above.
* @param node the node to insert
* @return node's predecessor
*/
private Node enq(final Node node) {
// 此处用到死循环，保证在并发情况下也能插入成功
for (;;) {
Node t = tail;
if (t == null) { // Must initialize
if (compareAndSetHead(new Node()))
tail = head;
} else {
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
从上述代码可看到，对于不同模式的入队操作只是在Node节点上的nextWaiter属性有所差异，其他部分完全相同。另外，对于等待节点的插入是通过尾插法来实现的，具体流程如下所示：
如上文介绍所说，通过
nextWaiter属性的不同取值来区分排他模式与共享模式。
- 通过
tail链接寻找到尾节点- 如果尾节点为
null，则初始化创建一个虚拟head节点，最后重复执行第1步 - 如果尾节点不为
null，则执行第2步
- 如果尾节点为
- 将新节点的
prev链接指向尾节点tail - 通过
CAS把新的节点设置为尾节点，如果失败则回到第1步继续执行，直到设置尾节点成功为止。 - 将原尾节点的
next链接指向新插入的节点，返回新插入的尾节点。
通过解读你会发现：即使新节点的prev链接指向了队尾节点也并不表示插入成功了，因为在并发情况下是有可能有多个线程同时执行了这一操作，而通过CAS操作只能保证有一个节点设置成功并执行next链接的设置，所以我们可以认为只有next链接设置成功后才算是节点插入成功。
这里对上文提及到的“当
next指向null并不代表当前节点就是队尾”的观点也作出了回应和解释。因为设置prev链接和next链接并不是原子操作，所以有可能在通过next链接判断后继节点时，另一个线程正在插入新节点且刚好执行到对prev链接进行赋值的操作。而对于这种情况，AQS是通过prev链接倒序遍历来进行一个double-check，这也相当于反向地利用了插入操作。
节点出队
在Node节点进入到等待队列后，就会进行自旋判断是否能成功获取到资源，并在一定条件下进入线程等待状态。而因为在排他模式与共享模式下节点出队的执行流程有所差异，所以在这里将分点进行阐述。
在传统
CLH锁中，因为仅仅是通过简单地自旋来判断当前是否能获取到资源并进行相应的出队操作，所以它无需其他唤醒操作。而AQS在自旋的基础上还加入了条件等待的机制来让线程进入等待状态，所以在使用上还是需要调用相应的unparkSuccessor方法来唤醒其后继节点，以让它继续执行自旋判断。
排他模式
所谓排他模式，即每次方法调用只能让一个节点获取到资源。
在排他模式中，在节点入队后会立即进行自旋判断是否能够获取资源，并在之后不断进行重试，直至获取成功为止（不考虑异常情况）。关于这部分代码并没有一个单独的方法进行包装，而是内嵌在acquire获取资源的方法中，下面笔者将与排他模式相关的几个acquire方法贴出来：
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
// 主要看这里，对于入队的节点立刻执行自旋+条件等待的出队操作
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
/**
* Acquires in exclusive uninterruptible mode for thread already in
* queue. Used by condition wait methods as well as acquire.
*
* @param node the node
* @param arg the acquire argument
* @return {@code true} if interrupted while waiting
*/
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
// 此处通过死循环保证了入队的节点肯定是能出队的，即重新去获取资源
for (;;) {
final Node p = node.predecessor();
// 此处判断了前驱节点为head节点，所以这里必然只能是公平策略
if (p == head && tryAcquire(arg)) {
// 将当前节点设置为head节点，并将原head节点出队
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
// 判断是否进入等待状态（获取资源失败时）
if (shouldParkAfterFailedAcquire(p, node) &&
// 暂停节点线程（由于这里是在一个死循环体内，所以在唤醒之后会继续获取资源，直到成功为止。）
parkAndCheckInterrupt())
// 此处中断是通过布尔型返回给调用方是否有中断
interrupted = true;
}
} finally {
// 如果发生错误（如抛出异常）则将对节点进行取消操作
if (failed)
cancelAcquire(node);
}
}
/**
* Acquires in exclusive interruptible mode.
* @param arg the acquire argument
*/
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)) {
// 将当前节点设置为head节点，并将原head节点出队
setHead(node);
p.next = null; // help GC
failed = false;
return;
}
// 判断是否进入等待状态（获取资源失败时）
if (shouldParkAfterFailedAcquire(p, node) &&
// 暂停节点线程（由于这里是在一个死循环体内，所以在唤醒之后会继续获取资源，直到成功为止。）
parkAndCheckInterrupt())
// 此处中断是抛出中断异常
throw new InterruptedException();
}
} finally {
// 如果发生错误（如抛出异常）则将对节点进行取消操作
if (failed)
cancelAcquire(node);
}
}
/**
* The number of nanoseconds for which it is faster to spin
* rather than to use timed park. A rough estimate suffices
* to improve responsiveness with very short timeouts.
*/
static final long spinForTimeoutThreshold = 1000L;
/**
* Acquires in exclusive timed mode.
*
* @param arg the acquire argument
* @param nanosTimeout max wait time
* @return {@code true} if acquired
*/
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();
// 此处判断了前驱节点为头节点，所以这里必然只能是公平策略
if (p == head && tryAcquire(arg)) {
// 将当前节点设置为head节点，并将原head节点出队
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);
}
}
在
doAcquireNanos中存在一个自旋优化，即nanosTimeout > spinForTimeoutThreshold，只有在超时时间大于这个阈值时才执行LockSupport.parkNanos方法暂停线程，否则直接执行自旋等待。官方给出的解释是：对于这个阈值之内的超时阻塞，自旋比停止调度要快。而在下文中与之类似的doAcquireSharedNanos方法也采取了类似的优化手段。
上述代码给出了三种类型的acquire方法，分别是acquireQueued、doAcquireInterruptibly和doAcquireNanos，这三个方法虽然看上去存在一些差异，但实际上它们核心的出队逻辑都是相同的，即通过自旋+等待队列。在这里笔者将其中节点出队的关键流程抽离了出来（伪代码）：
如果硬要说出
acquireQueued、doAcquireInterruptibly和doAcquireNanos的不同之处，可能就是在中断的处理、超时的处理上存在一些差别，但这些都不是理解核心流程的关键，此处可以先行忽略。
// 此处通过死循环保证了入队的节点肯定是能出队的，即重新去获取资源
for (;;) {
final Node p = node.predecessor();
// 此处判断了前驱节点为head节点，所以这里必然只能是公平策略
if (p == head && tryAcquire(arg)) {
// 将当前节点设置为head节点，并将原head节点出队
setHead(node);
p.next = null; // help GC
// ...
// ... 资源获取成功后返回 return val
}
// 判断是否超时
// ...
// 判断是否进入等待状态（获取资源失败时）
if (shouldParkAfterFailedAcquire(p, node) && ...){
// 暂停节点线程（由于这里是在一个死循环体内，所以在唤醒之后会继续获取资源，直到成功为止。）
parkThread()
}
// ...
// ... 中断处理
}
其中将当前节点设置为
head节点（即，将原head节点出队）的setHead方法逻辑如下所示：/** * Sets head of queue to be node, thus dequeuing. Called only by * acquire methods. Also nulls out unused fields for sake of GC * and to suppress unnecessary signals and traversals. * * @param node the node */ private void setHead(Node node) { head = node; // 当节点成为head节点时，thread变量存储的值就没用了（thread变量所存储的值用于执行线程等待和线程唤醒） node.thread = null; // head节点的前驱节点就是为null node.prev = null; }
即，具体的出队执行流程如下所示：
- 判断当前节点的前驱节点是否
head节点，如果不是则获取资源失败，跳到第4步 - 如果当前节点的前驱节点是
head节点，则尝试获取资源（通过tryAcquire方法定义），如果获取失败，跳到第4步 - 如果获取资源成功，则进行出队操作（将当前节点设置为
head节点，并让原head节点出队）- 将当前节点设置为
head节点（把当前节点的prev链接和thread变量设置为null） - 将原
head节点的next链接设置为null - 资源获取成功，返回结果值，执行结束
- 将当前节点设置为
- 判断当前节点是否进入等待状态，如果不是则跳回第
1步 - 如果当前节点可以进入等待状态，则调用方法暂停线程，直到线程被唤醒跳回第
1步（期间会判断中断状态）
- 在排他模式中获取资源的语义是通过
tryAcquire来定义的。- 通过
p == head来判断是否有资格获取资源，这必然是一种公平的策略，因为只有在前驱节点为head节点的情况下线程才有资格去获取资源。
然而，因为AQS会在一定条件下让队列节点进入线程等待状态，所以在节点执行完成后需要调用相应的unparkSuccessor方法来唤醒其后继节点。其中，在排他模式下是通过release的调用来触发unparkSuccessor方法的。
/**
* Releases in exclusive mode. Implemented by unblocking one or
* more threads if {@link #tryRelease} returns true.
* This method can be used to implement method {@link Lock#unlock}.
*
* @param arg the release argument. This value is conveyed to
* {@link #tryRelease} but is otherwise uninterpreted and
* can represent anything you like.
* @return the value returned from {@link #tryRelease}
*/
@ReservedStackAccess
public final boolean release(int arg) {
// 判断是否能释放资源
if (tryRelease(arg)) {
Node h = head;
// 判断head节点是否存在后继节点
if (h != null && h.waitStatus != 0)
// 唤醒后继节点
unparkSuccessor(h);
return true;
}
return false;
}
在release方法中首先会尝试去释放资源，如果成功才会去唤醒后继节点（如有）并返回true，否则直接返回false。其中，AQS用于唤醒后继节点的unparkSuccessor方法具体实现如下所示：
/**
* Wakes up node's successor, if one exists.
*
* @param node the node
*/
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);
/*
* Thread to unpark is held in successor, which is normally
* just the next node. But if cancelled or apparently null,
* traverse backwards from tail to find the actual
* non-cancelled successor.
*/
// 寻找后继节点
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);
}
对于unparkSuccessor方法其主要职责就是唤醒后继节点，其中主要有三个步骤：
- 尝试清理当前节点的状态，即设置为
0（成功与否都忽略） - 寻找当前节点的后继节点（如果节点被取消，则从队尾反向查询）
- 通过方法
LockSupport#unpark方法唤醒后继节点
最终，通过unparkSuccessor方法唤醒后继节点延续节点入队后的自旋重试获取资源，直到其获取成功为止（仅考虑成功的情况）。
总的来说，在排他模式下每个节点在获取资源失败后会进入等待状态，并在其前驱节点执行完成后被唤醒以继续尝试获取资源。
共享模式
所谓共享模式，即每次方法调用可以让多个节点获取到资源。
与排他模式类似，在共享模式中节点入队后会立即进行自旋判断是否能够获取资源，并在之后不断进行重试，直至获取成功为止（不考虑异常情况）。而关于这部分代码逻辑也是内嵌在了acquire获取资源的方法中，下面笔者将与共享模式相关的几个acquire方法贴出来：
/**
* Acquires in shared uninterruptible mode.
* @param arg the acquire argument
*/
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) {
// 将当前节点设置为head节点，并将原head节点出队，最后向后传播唤醒信号
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);
}
}
/**
* Acquires in shared interruptible mode.
* @param arg the acquire argument
*/
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
// 将节点加入等待队列
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) {
// 将当前节点设置为head节点，并将原head节点出队，最后向后传播唤醒信号
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
// 判断是否进入等待状态（获取资源失败时）
if (shouldParkAfterFailedAcquire(p, node) &&
// 暂停线程（由于这里是在一个死循环体内，所以在唤醒之后会继续获取资源，直到成功为止。）
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
// 如果发生错误（如抛出异常）则将对节点进行取消操作
if (failed)
cancelAcquire(node);
}
}
/**
* Acquires in shared timed mode.
*
* @param arg the acquire argument
* @param nanosTimeout max wait time
* @return {@code true} if acquired
*/
private boolean doAcquireSharedNanos(int arg, long nanosTimeout)
throws InterruptedException {
// 如果最大等待时间小于等于0则表示不等待，直接返回false（此处只有阻塞等待的节点才能进来）
if (nanosTimeout <= 0L)
return false;
// 计算最大等待时间
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) {
// 将当前节点设置为head节点，并将原head节点出队，最后向后传播唤醒信号
setHeadAndPropagate(node, r);
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);
}
}
上述代码同样给出了三种类型的acquire方法，分别是doAcquireShared、doAcquireSharedInterruptibly和doAcquireSharedNanos，它们的核心出队逻辑也是通过自旋+等待队列的方式，在这里笔者也将其中节点出队的关键流程抽离出来（伪代码）：
如果硬要说出
doAcquireShared、doAcquireSharedInterruptibly和doAcquireSharedNanos的不同之处，可能就是在中断的处理、超时的处理上存在一些差别，但这些都不是理解核心流程的关键，此处可以先行忽略。
// 此处通过死循环保证了入队的队列肯定是能出队的，即重新去获取资源
for (;;) {
final Node p = node.predecessor();
// 此处判断了前驱节点为头节点，所以这里必然只能是公平策略
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
// 将当前节点设置为head节点，并将原head节点出队，并向后传播唤醒信号（如有）
setHeadAndPropagate(node, r);
p.next = null; // help GC
// ...
// ... 中断处理
// ...
// ... 资源获取成功后返回 return val
}
}
// 判断是否超时
// ...
// 判断是否进入等待状态（获取资源失败时）
if (shouldParkAfterFailedAcquire(p, node) && ... ){
// 暂停线程（由于这里是在一个死循环体内，所以在唤醒之后会继续获取资源，直到成功为止。）
parkThread())
}
// ...
// ... 中断处理
}
其中将当前节点设置为
head节点（即，将原head节点出队）并向后传播唤醒信号（如有）的setHeadAndPropagate方法逻辑如下所示：/** * Sets head of queue, and checks if successor may be waiting * in shared mode, if so propagating if either propagate > 0 or * PROPAGATE status was set. * * @param node the node * @param propagate the return value from a tryAcquireShared */ private void setHeadAndPropagate(Node node, int propagate) { Node h = head; // Record old head for check below // 将当前节点设置为head节点（即，将原head节点出队），详情可阅读上文 setHead(node); // 判断是否向后传播唤醒信号 if (propagate > 0 || h == null || h.waitStatus < 0 || (h = head) == null || h.waitStatus < 0) { Node s = node.next; // 判断后继节点是否共享模式 if (s == null || s.isShared()) // 向后传播唤醒信号 doReleaseShared(); } }
即，具体的出队执行流程如下所示：
- 判断当前节点的前驱节点是否
head节点，如果不是则获取资源失败，跳到第4步 - 如果当前节点的前驱节点是头节点，则尝试获取资源（通过
tryAcquireShared方法定义），如果获取失败，跳到第4步 - 如果获取资源成功，则进行出队操作（将当前节点设置为
head节点，并让原head节点出队，然后向后传播唤醒信号）- 将当前节点设置为
head节点，并把节点的prev链接和thread变量设置为null - 满足一定条件下唤醒后继节点继续尝试获取资源（向后传播）
- 将原
head节点的next链接设置为null - 资源获取成功，返回结果值，执行结束
- 将当前节点设置为
- 判断当前节点是否进入等待状态，如果不是则跳回第
1步 - 如果当前节点可以进入等待状态，则调用方法暂停线程，直到线程被唤醒跳回第
1步（期间会判断中断状态）
- 在共享模式中获取资源的语义是通过
tryAcquireShared来定义的。- 通过
p == head来判断是否有资格获取资源，这必然是一种公平的策略，因为只有前驱节点是head节点的情况下线程才有资格去获取资源。
然而，因为AQS会在一定条件下让队列节点进入线程等待状态，所以在节点执行完成后需要调用相应的unparkSuccessor方法唤醒后继节点。其中，在共享模式下是通过releaseShared的调用来触发unparkSuccessor方法的。
/**
* Releases in shared mode. Implemented by unblocking one or more
* threads if {@link #tryReleaseShared} returns true.
*
* @param arg the release argument. This value is conveyed to
* {@link #tryReleaseShared} but is otherwise uninterpreted
* and can represent anything you like.
* @return the value returned from {@link #tryReleaseShared}
*/
@ReservedStackAccess
public final boolean releaseShared(int arg) {
// 判断是否能释放资源
if (tryReleaseShared(arg)) {
// 唤醒后继节点
doReleaseShared();
return true;
}
return false;
}
在releaseShared方法中首先会尝试去释放资源，如果成功才会去唤醒后继节点（如有）并返回true，否则直接返回false。其中，共享模式下唤醒后继节点的操作与排他模式存在一些差异，它在unparkSuccessor方法的基础上加入一些唤醒传播的逻辑，即doReleaseShared：
/**
* Release action for shared mode -- signals successor and ensures
* propagation. (Note: For exclusive mode, release just amounts
* to calling unparkSuccessor of head if it needs signal.)
*/
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) {
// 清理head节点的状态
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;
}
}
关于
doReleaseShared方法将waitStatus状态设置为PROPAGATE的原因可阅读下文《节点唤醒传播》小节，具体缘由在这里就不再继续展开了。
最终，doReleaseShared方法会不断的判断是否存在（需唤醒的）后继节点（通过SIGNAL状态判断），并在有的情况下通过unparkSuccessor方法将其唤醒。而如果被唤醒的后继节点成功获取到了资源则会一直向后传播唤醒信号，直至被唤醒的后继节点获取资源失败为止（通过比较head节点是否发生过变化）。
总的来说，在共享模式下每个节点在获取资源失败后会进入等待状态，在其前驱节点执行完成后被唤醒以继续尝试获取资源，并且在一定条件下不断的向后传播唤醒信号，直至被唤醒的后继节点获取资源失败为止。
本质上，无论是排他模式还是共享模式实现阻塞等待的机制都是基于CLH锁，只不过AQS在它的基础上加入了条件等待的策略。
扩展：实现细节
节点阻塞判断
根据上文的描述，在节点入队后就会不断的自旋判断是否能够获取到资源，并在一定条件下进入线程等待状态。而在AQS中统一（不论是排他模式还是共享模式）都是通过方法shouldParkAfterFailedAcquire来进行判断的（线程是否进入等待状态），这里笔者首先把相关代码贴出来：
/**
* Checks and updates status for a node that failed to acquire.
* Returns true if thread should block. This is the main signal
* control in all acquire loops. Requires that pred == node.prev.
*
* @param pred node's predecessor holding status
* @param node the node
* @return {@code true} if thread should block
*/
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;
}
shouldParkAfterFailedAcquire方法主要用于在资源获取失败后对节点状态进行检查和更新，并最终确认节点是否能进入等待状态。如果返回true表示节点可以进入等待状态，否则返回false表示不能进入等待状态，对于这种情况调用方法会在外层代码中继续调用这个方法进行重试，直到返回true为止。其中，对于shouldParkAfterFailedAcquire方法主要分为3种情况：
- 当前驱节点的
waitStatus状态为SIGNAL时，则表示节点可以进入等待状态，返回true - 当前驱节点的
waitStatus状态大于0时（节点被取消），则需重新将当前节点链接到有效的前驱节点（通过prev链接向前寻找），并返回false表示当前不能进入等待状态 - 当前驱节点的
waitStatus状态小于等于0时（除SIGNAL状态外），则需将节点更新为SIGNAL状态，并返回false表示当前不能进入等待状态
其中，对于第
3种通过CAS将前驱节点状态更新为SIGNAL的情况，由于它并不能保证一定能更新成功，所以这里还是会返回false表示当前节点不能进入等待状态。此时AQS会在外层代码中继续调用此方法，直到其返回true为止（如资源再次获取失败）。
即，只有在进入shouldParkAfterFailedAcquire方法时当前节点的前驱节点状态为SIGNAL时才能进入等待状态，其余情况则会在修复和更新操作后返回false表示目前不能进入等待状态。
结合上文对
SIGNAL状态所描述的语义和对unparkSuccessor方法所限定的调用条件，不难理解为什么只有前驱节点状态为SIGNAL时才能进入等待状态，简单来说就是只有当前head节点为SIGNAL时才会去唤醒其后继节点，所以此处必须保证只有其前驱节点状态被设置为SIGNAL时才能进入等待状态。
节点唤醒传播
在共享模式下实现的向后传播唤醒信号主要是通过doReleaseShared方法来实现，但是我们并不能直接对它进行调用，而是需要通过releaseShared方法和setHeadAndPropagate方法来间接触发doReleaseShared的调用（上文有所提及）。基于更深入理解向后传播唤醒信号的实现方案，在这里笔者将从releaseShared方法和setHeadAndPropagate方法进行展开分析。
-
对于
releaseShared方法，它主要用于在持有资源被释放成功后，对还存在等待资源的节点进行唤醒操作。/** * Releases in shared mode. Implemented by unblocking one or more * threads if {@link #tryReleaseShared} returns true. * * @param arg the release argument. This value is conveyed to * {@link #tryReleaseShared} but is otherwise uninterpreted * and can represent anything you like. * @return the value returned from {@link #tryReleaseShared} */ @ReservedStackAccess public final boolean releaseShared(int arg) { // 判断是否能释放资源 if (tryReleaseShared(arg)) { // 向后传播唤醒信号 doReleaseShared(); return true; } return false; } -
对于
setHeadAndPropagate方法，它主要用于在head节点成功获取资源后，在可能还存在资源的情况下向后传播唤醒信号。/** * Sets head of queue, and checks if successor may be waiting * in shared mode, if so propagating if either propagate > 0 or * PROPAGATE status was set. * * @param node the node * @param propagate the return value from a tryAcquireShared */ private void setHeadAndPropagate(Node node, int propagate) { Node h = head; // Record old head for check below // 将当前节点设置为head节点（即，将原head节点出队），详情可阅读上文 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(); } }
基于共享模式的设计理念，只要存在有剩余资源的可能性就会执行向后传播唤醒信号，让更多的节点（线程）能够获取到资源得以执行。
对于releaseShared方法，在释放资源成功后便会向后传播唤醒信号让后继节点获取刚刚被释放的资源，这其中的逻辑并不难理解。与之相对，setHeadAndPropagate方法向后传播唤醒信号的判断条件则是比较难理解了，这里笔者把这部分代码单独拎了出来：
/*
* 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)
从上述代码能得出只要满足其中一个条件就能执行doReleaseShared方法进行唤醒信号地向后传播，这里笔者把这5个条件拆分开并逐个进行分析：
propagate > 0，在propagate > 0的情况下允许向后传播h == null，在原head节点为null的情况下允许向后传播h.waitStatus < 0，在原head节点的waitStatus < 0的情况下允许向后传播(h = head) == null，在新head节点为null的情况下允许向后传播h.waitStatus < 0，在新head节点的waitStatus < 0的情况下允许向后传播
需要注意，从上至下每一个条件都是基于其上一个条件的反向逻辑。
对于第1点在propagate > 0的情况下允许向后传播这个条件应该比较容易理解，因为根据共享模式的定义当存在剩余资源propagate时是可以向后传播的。而对于第2、3、4、5点理解起来并没有那么直观，因为它们的成立意味着propagate <= 0，这并不符合共享模式的定义。此处，我们需要换一种思路来分析，即在并发条件下propagate参数值可能并不准确、head节点的变更可能处于临界状态等，如果仅仅判断propagate则可能会在存在资源的情况下停止了唤醒信号的传播，进而导致等待节点永远不会被唤醒。
也就是说，当在doAcquireShared方法中线程获取资源成功，并将propagate=0参数传入到setHeadAndPropagate方法时，如果当前节点还存在后继节点（通过waitStatus < 0表示其存在后继节点）则会保守地进行一次传播操作，因为此时可能存在曾经的head节点释放了持有的资源。
另外，在确定可以向后传播后会进一步判断其后继节点是否为共享模式，此时如果其后继节点为null同样也会触发一次doReleaseShared方法，这是因为当next==null时有可能存在节点刚好执行入队操作（next==null并不表示其就是队尾，具体缘由可以回顾上文）。即，此处依然采用了保守策略执行了一次向后传播。
综上所述，在并发条件下这些保守的边界判断能保证程序更好、更快的执行，就像注释上说的：“这些检查是保守的，可能会引起不必要的唤醒操作”。
在经过上述一系列的判断后我们就开始执行doReleaseShared方法，即接下来我们就doReleaseShared方法是如何执行向后传播唤醒信号展开分析。
/**
* Release action for shared mode -- signals successor and ensures
* propagation. (Note: For exclusive mode, release just amounts
* to calling unparkSuccessor of head if it needs signal.)
*/
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;
// 如果head节点状态为SIGNAL=-1（表示存在后继节点），则将其状态从SIGNAL(-1)设置为INIT(0)
if (ws == Node.SIGNAL) {
// 清理head节点状态
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
// 唤醒后继节点
unparkSuccessor(h);
}
// 如果head节点状态为INIT=0（表示不存在可唤醒的后继节点），则将其状态从INIT(0)设置为PROPAGATE(-3)
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
// 如果原head节点与新head节点相同，则表示唤醒的后继节点获取资源失败（或者不存在需唤醒的后继节点），退出循环
if (h == head) // loop if head changed
break;
}
}
在这里笔者把doReleaseShared方法相应的执行步骤归纳了出来：
- 判断当前
head节点是否存在后继节点，否则跳到第3步 - 判断当前
head节点的状态是否符合传播条件，否则跳到第3步- 如果
head节点状态为SIGNAL=-1（表示存在可唤醒的后继节点），则将其状态从SIGNAL(-1)设置为INIT(0)，并唤醒其后继节点 - 如果
head节点状态为INIT=0（表示不存在可唤醒的后继节点），则将其状态从INIT(0)设置为PROPAGATE(-3)
- 如果
- 判断
head节点是否发生过变化（通过比较h == head表达式，其中h为开始执行时head节点快照，head则是最新的head节点）- 如果发生过变化，跳回到第
1步 - 如果没有发生过变化，则方法执行结束
- 如果发生过变化，跳回到第
后继节点（被唤醒的）在成功获取资源后，
head节点就会发生改变（即h != head），这种情况方法是会继续执行的。而如果不存在后继节点或者后继节点获取资源失败，则head节点并没有发生改变（即h == head），这种情况是跳出循环终止执行的。这里充分地体现了共享模式的本质，即当存在后继节点时会一直往后传播唤醒信号，直到唤醒的后继节点获取资源失败为止（一次调用可唤醒多个等待节点）。
在此处就引出了一个问题，为什么在共享模式下需要将状态0设置为PROPAGATE(-3)呢？或者换个说法，为什么需要存在PROPAGATE(-3)这个状态呢？
我们不妨先来看看PROPAGATE(-3)的状态说明：“A releaseShared should be propagated to other nodes. This is set (for head node only) in doReleaseShared to ensure propagation continues, even if other operations have since intervened.”。简单来说就是releaseShared的调用应该向后传播唤醒信号，而将状态设置为PROPAGATE(-3)可以保证传播顺利的进行。换句话说，如果没有这个PROPAGATE(-3)状态就会导致传播中断进而发生异常，但是具体导致异常的原因在这里却并没有十分清晰的描述。
基于此笔者在网上查阅了相关的资料并最终找到了答案，即在JDK6的某个版本（修复前版本）中存在这么一个问题：“并发执行releaseShared方法会导致部分等待节点（线程）没有被唤醒”，为此Doug Lea大佬提交了一个commit修复了这个问题，即《6801020: Concurrent Semaphore release may cause some require thread not signaled》，其中主要的解决方案在commit中是这样描述的：“Introduce PROPAGATE waitStatus”，即通过引入PROPAGATE状态。
其中
Semaphore是基于AQS共享模式实现的一个信号量工具类。
既然已经知道了引入PROPAGATE状态的原因，那么这里我们就顺着Doug Lea大佬提供的思路将所产生的BUG复现出来，然后再通过引入PROPAGATE状态来将BUG解决了。
首先这里将触发BUG的单元测试贴了出来（为了便于理解，笔者将其中非核心的部分剔除了）：
// 单元测试
Semaphore sem = new Semaphore(0, fair);
Runnable blocker = ()->sem.acquire();
Runnable signaller = ()->sem.release();
Thread b1 = new Thread(blocker);
Thread b2 = new Thread(blocker);
Thread s1 = new Thread(signaller);
Thread s2 = new Thread(signaller);
Thread[] threads = { b1, b2, s1, s2 };
for (Thread thread : threads)
thread.start();
for (Thread thread : threads) {
thread.join(60 * 1000);
}
if (sem.availablePermits() != 0)
throw new Error(String.valueOf(sem.availablePermits()));
if (sem.hasQueuedThreads())
throw new Error(String.valueOf(sem.hasQueuedThreads()));
if (sem.getQueueLength() != 0)
throw new Error(String.valueOf(sem.getQueueLength()));
接着，我们再来看看修复前引发BUG的代码，分别是setHeadAndPropagate方法和releaseShared方法:
// 修复前
private void setHeadAndPropagate(Node node, int propagate) {
setHead(node);
if (propagate > 0 && node.waitStatus != 0) {
Node s = node.next;
if (s == null || s.isShared())
unparkSuccessor(node);
}
}
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
在旧版本的setHeadAndPropagate方法和releaseShared方法中并不存在对doReleaseShared方法的调用，而是在判断propagate和waitStatus（不存在PROPAGATE状态）后直接调用unparkSuccessor方法执行唤醒操作。
至此，在基于上述旧版本的代码中笔者作出如下假设：
-
首先线程
b1和b2分别执行了Semaphore#acquire方法（即，执行AQS#doAcquireShared方法），而因为初始化信号量为0（即不存在资源），所以两次获取资源都失败了，并分别进入到等待队列中。如下图所示：step1：线程b1和b2执行了AQS#doAcquireShared方法获取资源失败，构造为Node节点进入等待队列 state = 0 head(虚拟节点) +---------+ +----------+ +----------+ | +------>+ +------->+ | | Node1 | | Node2(b1)| | Node3(b2)| | +<------+ +<-------+ | +---------+ +----------+ +----------+ -
然后线程
s1执行了Semaphore#release方法（即，执行AQS#releaseShared方法），此时线程s1调用AQS#tryReleaseShared方法释放了1个资源（资源+1），并因为head（Node1）节点的waitStatus==SIGNAL（不等于0），所以接着唤醒后继节点Node2(b1)并将head（Node1）节点的waitStatus设置为0。如下图所示：step2：线程s1执行了AQS#tryReleaseShared方法释放1个资源，并唤醒后继节点Node2(b1) state = 1 head(虚拟节点) executing +---------+ +----------+ +----------+ | +------>+ +------->+ | | Node1 | | Node2(b1)| | Node3(b2)| | +<------+ +<-------+ | +---------+ +----------+ +----------+ -
然后线程
b1执行AQS#tryAcquireShared方法持有1个资源（资源-1），并进入到AQS#setHeadAndPropagate方法（开始处）。如下图所示：step3：线程b1执行了AQS#tryAcquireShared方法持有1个资源 state = 0 head(虚拟节点) executing +---------+ +----------+ +----------+ | +------>+ +------->+ | | Node1 | | Node2(b1)| | Node3(b2)| | +<------+ +<-------+ | +---------+ +----------+ +----------+// 修复前 private void setHeadAndPropagate(Node node, int propagate) { // o <- 此处为临界点 // 此时head=Node1，node=Node2(b1)，propagate=0 setHead(node); if (propagate > 0 && node.waitStatus != 0) { Node s = node.next; if (s == null || s.isShared()) unparkSuccessor(node); } } -
接着线程
s2执行了Semaphore#release方法（即，执行AQS#releaseShared方法），此时线程s2调用AQS#tryReleaseShared方法释放了1个资源（资源+1），但是因为head节点仍然为Node1，并不符合唤醒后继节点的条件（waitStatus==0），所以直接执行结束。如下图所示：step4：线程s2执行了AQS#tryReleaseShared方法释放1个资源，但不唤醒后继节点 state = 1 head(虚拟节点) executing +---------+ +----------+ +----------+ | +------>+ +------->+ | | Node1 | | Node2(b1)| | Node3(b2)| | +<------+ +<-------+ | +---------+ +----------+ +----------+ -
最后线程
b1被设置为head节点，但是由于传入的propagate（快照）为0并不符合唤醒后继节点的条件，所以直接执行结束，如下图所示：step5：线程b1被设置为head节点，但不唤醒后继节点 state = 1 head +---------+ +----------+ +----------+ | +- ->+ +------->+ | | Node1 | | Node2(b1)| | Node3(b2)| | +< --+ +<-------+ | +---------+ +----------+ +----------+
最终结果，在还剩余1个信号量的情况下停止了唤醒信号的传播，进而导致节点Node3(b2)永远停留在队列中（不考虑中断）。
为此，Doug Lea大佬引入了PROPAGATE(-3)状态以解决该问题，并在setHeadAndPropagate方法和releaseShared方法中作出相应的修改，最终得以解决。具体如下：
// 修改后
private void setHeadAndPropagate(Node node, long propagate) {
Node h = head;
setHead(node);
if (propagate > 0 || h == null || h.waitStatus < 0) {
Node s = node.next;
if (s == null || s.isShared())
doReleaseShared();
}
}
public final boolean releaseShared(long arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
private void doReleaseShared() {
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;
}
}
下面我们再按照上述假设重新执行一遍流程：
-
首先线程
b1和b2分别执行了Semaphore#acquire方法（即，执行AQS#doAcquireShared方法），而因为初始化信号量为0（即不存在资源），所以两次获取资源都失败了，并分别进入到等待队列中。如下图所示：step1：线程b1和b2执行了AQS#doAcquireShared方法获取资源失败，构造为Node节点进入等待队列 state = 0 head(虚拟节点) +---------+ +----------+ +----------+ | +------>+ +------->+ | | Node1 | | Node2(b1)| | Node3(b2)| | +<------+ +<-------+ | +---------+ +----------+ +----------+ -
然后线程
s1执行了Semaphore#release方法（即，执行AQS#releaseShared方法），此时线程s1调用AQS#tryReleaseShared释放了1个资源（资源+1），因此可以进一步执行doReleaseShared方法。在doReleaseShared方法中，由于head（Node1）节点的waitStatus==SIGNAL，所以将head（Node1）节点的waitStatus设置为0并且唤醒后继节点Node2(b1)。如下图所示：step2：线程s1调用AQS#tryReleaseShared方法释放1个资源，并调用doReleaseShared方法唤醒后继节点Node2(b1) state = 1 head(虚拟节点) executing +---------+ +----------+ +----------+ | +------>+ +------->+ | | Node1 | | Node2(b1)| | Node3(b2)| | +<------+ +<-------+ | +---------+ +----------+ +----------+ -
然后线程
b1执行AQS#tryAcquireShared方法持有1个资源（资源-1），并进入到AQS#setHeadAndPropagate方法（开始处）。如下图所示：step3：线程b1执行AQS#tryAcquireShared方法持有1个资源 state = 0 head(虚拟节点) executing +---------+ +----------+ +----------+ | +------>+ +------->+ | | Node1 | | Node2(b1)| | Node3(b2)| | +<------+ +<-------+ | +---------+ +----------+ +----------+// 修改后 private void setHeadAndPropagate(Node node, long propagate) { Node h = head; // o <- 此处为临界点 // 此时head=Node1，node=Node2(b1)，propagate=0 setHead(node); if (propagate > 0 || h == null || h.waitStatus < 0) { Node s = node.next; if (s == null || s.isShared()) doReleaseShared(); } } -
接着线程
s2执行了Semaphore#release方法（即，执行AQS#releaseShared方法），此时线程s2调用AQS#tryReleaseShared方法释放了1个资源（资源+1），因此可以进一步执行doReleaseShared方法。在doReleaseShared方法中，由于head（Node1）节点仍然为Node1，所以此处条件判断后执行了分支2（waitStatus==0）将head（Node1）节点的状态从0设置为PROPAGATE(-3)，最后结束方法的执行（由于head节点并没有发生变化）。如下图所示：step4：线程s2调用AQS#tryReleaseShared方法释放了1个资源，将当前head节点状态设置为PROPAGATE(-3) state = 1 head(虚拟节点) executing +---------+ +----------+ +----------+ | +------>+ +------->+ | | Node1 | | Node2(b1)| | Node3(b2)| | +<------+ +<-------+ | +---------+ +----------+ +----------+ -
接着线程
b1被设置为head节点，而由于原head节点（即，变量h）状态被设置为PROPAGATE(-3)符合条件waitStatus < 0，因此可以进一步执行进入到doReleaseShared方法。在doReleaseShared方法中，由于head（Node2）节点的waitStatus==SIGNAL，所以将head（Node2）节点的waitStatus设置为0并且唤醒后继节点Node3(b2)。如下图所示：step5：线程b1被设置为head节点，并唤醒后继节点 state = 1 head executing +---------+ +----------+ +----------+ | +- ->+ +------->+ | | Node1 | | Node2(b1)| | Node3(b2)| | +< --+ +<-------+ | +---------+ +----------+ +----------+ -
最后线程
b2执行AQS#tryAcquireShared方法持有1个资源（资源-1），并进入到AQS#setHeadAndPropagate方法，将节点Node3(b2)设置为head节点。最终因为它不存在后继节点（即，propagate==0 && waitStatus==0），所以直接执行结束。如下图所示：step6：线程b2被设置为head节点，并结束执行 state = 0 head +---------+ +----------+ +----------+ | +- ->+ +-- ->+ | | Node1 | | Node2(b1)| | Node3(b2)| | +< --+ +<- --+ | +---------+ +----------+ +----------+
至此，AQS在引入PROPAGATE(-3)状态后避免了传播中断问题的再次发生。
思考：在分析
PROPAGATE作用的时候发现它只有在doReleaseShared方法中才会被设置，但对于它的比较却没有使用过一条特殊的等式，基本是通过waitStatus < 0来判断的，可以更直接地说它是为setHeadAndPropagate方法的条件判断而生的。这不禁让我陷入了沉思，状态PROPAGATE是否必要的呢？或者说在不考虑语义的前提下状态PROPAGATE是否能用SIGNAL来代替呢？
阻塞原理
在对AQS的应用中，对于等待队列的一些底层方法（如doAcquire、doAcquireShared等）一般是不会直接进行调用的，而是通过其上层的包装方法间接地对其进行调用。接下来我们就从上层的维度来分析其线程阻塞的原理。
排他阻塞
在AQS中，如果要在排他模式下执行阻塞式获取资源则可以通过调用acquire或acquireInterruptibly方法来达到目的，即在资源充足的情况下调用acquire或acquireInterruptibly方法获得资源并执行相应的逻辑；而在资源不足的情况下调用acquire或acquireInterruptibly方法则会进入阻塞状态。
/**
* Acquires in exclusive mode, ignoring interrupts. Implemented
* by invoking at least once {@link #tryAcquire},
* returning on success. Otherwise the thread is queued, possibly
* repeatedly blocking and unblocking, invoking {@link
* #tryAcquire} until success. This method can be used
* to implement method {@link Lock#lock}.
*
* @param arg the acquire argument. This value is conveyed to
* {@link #tryAcquire} but is otherwise uninterpreted and
* can represent anything you like.
*/
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
/**
* Acquires in exclusive mode, aborting if interrupted.
* Implemented by first checking interrupt status, then invoking
* at least once {@link #tryAcquire}, returning on
* success. Otherwise the thread is queued, possibly repeatedly
* blocking and unblocking, invoking {@link #tryAcquire}
* until success or the thread is interrupted. This method can be
* used to implement method {@link Lock#lockInterruptibly}.
*
* @param arg the acquire argument. This value is conveyed to
* {@link #tryAcquire} but is otherwise uninterpreted and
* can represent anything you like.
* @throws InterruptedException if the current thread is interrupted
*/
public final void acquireInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (!tryAcquire(arg))
doAcquireInterruptibly(arg);
}
关于acquire或acquireInterruptibly方法的核心流程是相同的，即首先会通过tryAcquire方法尝试获取资源（具体的获取语义则需要使用者自定义），如果资源被获取成功则直接返回；而如果资源获取失败，则通过acquireQueued和doAcquireInterruptibly等方法将线程构造为Node节点插入到等待队列中进行阻塞等待。
在获得资源的线程在执行完成后，则需要调用相应的release方法释放持有的资源，而在排他模式下会唤醒一个后继的等待线程继续尝试获取资源。
/**
* Releases in exclusive mode. Implemented by unblocking one or
* more threads if {@link #tryRelease} returns true.
* This method can be used to implement method {@link Lock#unlock}.
*
* @param arg the release argument. This value is conveyed to
* {@link #tryRelease} but is otherwise uninterpreted and
* can represent anything you like.
* @return the value returned from {@link #tryRelease}
*/
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
在release方法中，它首先会通过tryRelease方法尝试释放资源（具体的释放语义则需要使用者自定义），如果资源被释放成功则唤醒后继节点（如条件符合）并返回true，否则直接返回false退出执行。
共享阻塞
如果要在共享模式下执行阻塞式获取资源则可以通过调用acquireShared方法来达到目的，即在资源充足的情况下调用doAcquireShared方法获得资源，并在条件符合下继续往后传播唤醒信号以让更多的等待线程可以获得资源，最终执行相应的逻辑；而如果资源不足的情况下调用doAcquireShared方法同样会进入阻塞状态。
/**
* Acquires in shared mode, ignoring interrupts. Implemented by
* first invoking at least once {@link #tryAcquireShared},
* returning on success. Otherwise the thread is queued, possibly
* repeatedly blocking and unblocking, invoking {@link
* #tryAcquireShared} until success.
*
* @param arg the acquire argument. This value is conveyed to
* {@link #tryAcquireShared} but is otherwise uninterpreted
* and can represent anything you like.
*/
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
在acquireShared方法中，它首先会通过tryAcquireShared方法尝试获取资源（具体的获取语义则需要使用者自定义），如果资源获取成功则直接返回；而如果资源获取失败，则通过doAcquireShared方法将线程构造为Node节点插入到等待队列中进行阻塞等待。
在获得资源的线程在执行完成后，则需要调用相应的releaseShared方法释放持有的资源，而在共享模式下会唤醒一个后继的等待线程继续尝试获取资源，并在条件符合下继续往后传播唤醒信号以让更多的等待线程可以获得资源，最终执行相应的逻辑。
/**
* Releases in shared mode. Implemented by unblocking one or more
* threads if {@link #tryReleaseShared} returns true.
*
* @param arg the release argument. This value is conveyed to
* {@link #tryReleaseShared} but is otherwise uninterpreted
* and can represent anything you like.
* @return the value returned from {@link #tryReleaseShared}
*/
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
在releaseShared方法中，它首先会通过tryReleaseShared方法尝试释放资源（具体的释放语义则需要使用者自定义），如果资源被释放成功则唤醒后继节点（如条件符合，会一直往后传播唤醒信号）并返回true，否则直接返回false退出执行。
此处需要注意，虽然
acquireQueued、doAcquireInterruptibly和doAcquireShared等方法的阻塞机制都是公平的，但是如果在实现同步器/锁时不加以注意则可能会得到一个非公平的同步器/锁。这是因为在每次调用acquire/acquireShared方法时首先都会先执行tryAcquire/tryAcquireShared方法，如果它执行成功则表示拿到了资源，而不用再陷入阻塞状态，即抢占了等待时间最长的线程。所以，如果需要实现公平的同步器/锁，则需要在tryAcquire/tryAcquireShared方法上做文章了。
条件队列
AQS不但支持同步阻塞机制，而且还支持条件等待机制。其中，条件等待机制是通过条件变量condition variables（也被称为条件队列condition queues）来实现的，而为了与上文相呼应，下文统一用条件队列condition queues来命名。
条件变量
condition variables的概念来自于Monitor，它表示一个与mutex相关联的节点（线程）队列，队列中的每个节点（线程）都会在不占用所关联的mutex的情况下进入到等待状态（以让其他节点（线程）可以获取到mutex），直到某个条件成立为止。
Condition接口
对于AQS的条件队列condition queues是基于Condition接口来实现的，下面笔者将Condition接口相关的注释说明贴了出来：
/**
* {@code Condition} factors out the {@code Object} monitor
* methods ({@link Object#wait() wait}, {@link Object#notify notify}
* and {@link Object#notifyAll notifyAll}) into distinct objects to
* give the effect of having multiple wait-sets per object, by
* combining them with the use of arbitrary {@link Lock} implementations.
* Where a {@code Lock} replaces the use of {@code synchronized} methods
* and statements, a {@code Condition} replaces the use of the Object
* monitor methods.
*
* <p>Conditions (also known as <em>condition queues</em> or
* <em>condition variables</em>) provide a means for one thread to
* suspend execution (to "wait") until notified by another
* thread that some state condition may now be true. Because access
* to this shared state information occurs in different threads, it
* must be protected, so a lock of some form is associated with the
* condition. The key property that waiting for a condition provides
* is that it <em>atomically</em> releases the associated lock and
* suspends the current thread, just like {@code Object.wait}.
*
* <p>A {@code Condition} instance is intrinsically bound to a lock.
* To obtain a {@code Condition} instance for a particular {@link Lock}
* instance use its {@link Lock#newCondition newCondition()} method.
*
* ...
*
* <p>A {@code Condition} implementation can provide behavior and semantics
* that is
* different from that of the {@code Object} monitor methods, such as
* guaranteed ordering for notifications, or not requiring a lock to be held
* when performing notifications.
* If an implementation provides such specialized semantics then the
* implementation must document those semantics.
*
* <p>Note that {@code Condition} instances are just normal objects and can
* themselves be used as the target in a {@code synchronized} statement,
* and can have their own monitor {@link Object#wait wait} and
* {@link Object#notify notification} methods invoked.
* Acquiring the monitor lock of a {@code Condition} instance, or using its
* monitor methods, has no specified relationship with acquiring the
* {@link Lock} associated with that {@code Condition} or the use of its
* {@linkplain #await waiting} and {@linkplain #signal signalling} methods.
* It is recommended that to avoid confusion you never use {@code Condition}
* instances in this way, except perhaps within their own implementation.
*
* <p>Except where noted, passing a {@code null} value for any parameter
* will result in a {@link NullPointerException} being thrown.
*
* <h3>Implementation Considerations</h3>
*
* <p>When waiting upon a {@code Condition}, a "<em>spurious
* wakeup</em>" is permitted to occur, in
* general, as a concession to the underlying platform semantics.
* This has little practical impact on most application programs as a
* {@code Condition} should always be waited upon in a loop, testing
* the state predicate that is being waited for. An implementation is
* free to remove the possibility of spurious wakeups but it is
* recommended that applications programmers always assume that they can
* occur and so always wait in a loop.
*
* <p>The three forms of condition waiting
* (interruptible, non-interruptible, and timed) may differ in their ease of
* implementation on some platforms and in their performance characteristics.
* In particular, it may be difficult to provide these features and maintain
* specific semantics such as ordering guarantees.
* Further, the ability to interrupt the actual suspension of the thread may
* not always be feasible to implement on all platforms.
*
* <p>Consequently, an implementation is not required to define exactly the
* same guarantees or semantics for all three forms of waiting, nor is it
* required to support interruption of the actual suspension of the thread.
*
* <p>An implementation is required to
* clearly document the semantics and guarantees provided by each of the
* waiting methods, and when an implementation does support interruption of
* thread suspension then it must obey the interruption semantics as defined
* in this interface.
*
* <p>As interruption generally implies cancellation, and checks for
* interruption are often infrequent, an implementation can favor responding
* to an interrupt over normal method return. This is true even if it can be
* shown that the interrupt occurred after another action that may have
* unblocked the thread. An implementation should document this behavior.
*
* @since 1.5
* @author Doug Lea
*/
public interface Condition {
}
此处笔者从中提取出几点关键点：
Condition实例需要绑定一个Lock，一般在通过Lcok#newCondition方法创建Condition实例时进行绑定。Condition提供了一种方式让我们可以释放所关联的Lock并暂停线程的执行，直到在条件符合的情况下被其他线程通知，即条件队列。
在
Java中，Object#monitor（即Object#wait()与Object#notify()）具有与Condition类似的功能，只不过Object#monitor需要与synchronized联合使用，而Condition则需要与Lock联合使用。另外，最终实现的Condition在行为上和在语义上不一定与Object#monitor完全一样（例如顺序性保证、锁持有保证等），这依赖于具体的实现。
而在Condition中主要定义了两类方法，分别是用于让线程进入等待状态的await类方法和用于通知等待线程的signal类方法。下面我们来看看Condition是如何对它们进行定义的：
-
await类方法/** * Causes the current thread to wait until it is signalled or * {@linkplain Thread#interrupt interrupted}. * * <p>The lock associated with this {@code Condition} is atomically * released and the current thread becomes disabled for thread scheduling * purposes and lies dormant until <em>one</em> of four things happens: * <ul> * <li>Some other thread invokes the {@link #signal} method for this * {@code Condition} and the current thread happens to be chosen as the * thread to be awakened; or * <li>Some other thread invokes the {@link #signalAll} method for this * {@code Condition}; or * <li>Some other thread {@linkplain Thread#interrupt interrupts} the * current thread, and interruption of thread suspension is supported; or * <li>A "<em>spurious wakeup</em>" occurs. * </ul> * * <p>In all cases, before this method can return the current thread must * re-acquire the lock associated with this condition. When the * thread returns it is <em>guaranteed</em> to hold this lock. * * <p>If the current thread: * <ul> * <li>has its interrupted status set on entry to this method; or * <li>is {@linkplain Thread#interrupt interrupted} while waiting * and interruption of thread suspension is supported, * </ul> * then {@link InterruptedException} is thrown and the current thread's * interrupted status is cleared. It is not specified, in the first * case, whether or not the test for interruption occurs before the lock * is released. * * <p><b>Implementation Considerations</b> * * <p>The current thread is assumed to hold the lock associated with this * {@code Condition} when this method is called. * It is up to the implementation to determine if this is * the case and if not, how to respond. Typically, an exception will be * thrown (such as {@link IllegalMonitorStateException}) and the * implementation must document that fact. * * <p>An implementation can favor responding to an interrupt over normal * method return in response to a signal. In that case the implementation * must ensure that the signal is redirected to another waiting thread, if * there is one. * * @throws InterruptedException if the current thread is interrupted * (and interruption of thread suspension is supported) */ void await() throws InterruptedException;因为对于具有中断、非中断、超时等功能的
await方法它们的核心流程基本上是相同的，所以在这里就不再展开了。对
await的定义官方是这样描述的：执行await将使得与Condition相关联的Lock自动地释放并让当前线程进入等待状态，直到线程被通知或被中断唤醒。其中，存在以下几种方法可以让等待线程被唤醒：- 其他线程执行
Condition#signal()，并且当前线程刚好被选中唤醒 - 其他线程执行
Condition#signalAll()，唤醒所有等待线程 - 其他线程执行
Thread#interrupt()中断当前线程（如果是方法awaitUninterruptibly则不会存这一条唤醒规则） - 当前线程等待超时被唤醒（如果方法有设置超时时间，例如
awaitNanos、awaitUntil） - 当前线程被虚假唤醒
需要注意的是，在等待线程被唤醒后需要成功获取
Condition所关联的Lock后才能从（使线程进入等待状态的）await方法返回，否则线程将继续等待（保证await返回时持有了Lock） - 其他线程执行
-
signal类方法/** * Wakes up one waiting thread. * * <p>If any threads are waiting on this condition then one * is selected for waking up. That thread must then re-acquire the * lock before returning from {@code await}. * * <p><b>Implementation Considerations</b> * * <p>An implementation may (and typically does) require that the * current thread hold the lock associated with this {@code * Condition} when this method is called. Implementations must * document this precondition and any actions taken if the lock is * not held. Typically, an exception such as {@link * IllegalMonitorStateException} will be thrown. */ void signal(); /** * Wakes up all waiting threads. * * <p>If any threads are waiting on this condition then they are * all woken up. Each thread must re-acquire the lock before it can * return from {@code await}. * * <p><b>Implementation Considerations</b> * * <p>An implementation may (and typically does) require that the * current thread hold the lock associated with this {@code * Condition} when this method is called. Implementations must * document this precondition and any actions taken if the lock is * not held. Typically, an exception such as {@link * IllegalMonitorStateException} will be thrown. */ void signalAll();在
await的方法定义中，我们已经对signal方法和signalAll方法的作用进行了阐述，即signal方法用于唤醒一个等待线程，而signalAll方法则用于唤醒所有等待线程，但是无论是通过signal方法还是signalAll方法唤醒的线程都必须重新获取Lock才能使线程从（使线程进入等待状态的）await方法返回。
ConditionObject实现
在了解完AQS对条件队列的定义后，接下来我们一起来看看其具体的实现ConditionObject。
public abstract class AbstractQueuedSynchronizer
extends AbstractOwnableSynchronizer
implements java.io.Serializable {
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;
public ConditionObject() {}
}
}
ConditionObject在实现条件队列时复用了等待队列的数据结构Node，并通过额外的字段nextWaiter对每个Node节点进行链接，然后再通过ConditionObject中的firstWaiter和lastWaiter链接分别指向其头节点与尾节点，最终形成条件队列。具体结构如下图所示：
条件队列是通过
nextWaiter链接进行实现的单向链表，而等待队列则是通过next链接和prev链接实现的双向链表。
condition队列：
Node Node Node Node
+---------------+ +---------------+ +---------------+ +---------------+
| thread | | thread | | thread | | thread |
| waitStatus | | waitStatus | | waitStatus | | waitStatus |
| prev | | prev | | prev | | prev |
| next | | next | | next | | next |
| | | | | | | |
| | | | | | | |
| nextWaiter+--------->+ nextWaiter+---------->+ nextWaiter+--------->+ nextWaiter+--------->null
+-------+-------+ +---------------+ +---------------+ +--------+------+
^ ^
| |
firstWaiter+------+ +-------------+lastWaiter
在了解完ConditionObject的数据结构后，我们再来看看它是如何实现在Condition上定义的await类方法和signal类方法。
await等待
在持有Lock的情况下，执行await方法将会导致线程进入条件等待状态，并且释放掉持有的Lock。下面笔者将await方法相关代码贴了出来：
使用
ConditionObject的前提条件是需要持有Lock，也就说只有在持有Lock的情况下才能调用await方法。
public class ConditionObject implements Condition, java.io Serializable {
/**
* Implements interruptible condition wait.
* <ol>
* <li> If current thread is interrupted, throw InterruptedException.
* <li> Save lock state returned by {@link #getState}.
* <li> Invoke {@link #release} with saved state as argument,
* throwing IllegalMonitorStateException if it fails.
* <li> Block until signalled or interrupted.
* <li> Reacquire by invoking specialized version of
* {@link #acquire} with saved state as argument.
* <li> If interrupted while blocked in step 4, throw InterruptedException.
* </ol>
*/
public final void await() throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
// 1. 把当前节点（线程）链接到条件队列
Node node = addConditionWaiter();
// 2. 将当前节点（线程）持有的资源释放掉，并唤醒其后继节点（如有）
int savedState = fullyRelease(node);
int interruptMode = 0;
// 3. 自旋判断当前节点（线程）是否在等待队列
while (!isOnSyncQueue(node)) {
LockSupport.park(this);
// 判断是否存在中断，如存在则跳出循环
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
// 4. 在等待队列中对于当前节点（线程）执行节点出队操作
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
// 5. 将条件队列中的取消节点移除（执行清理操作）
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
// 6. 对等待过程中发生的中断进行处理（重新标记中断/抛出中断异常）
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}
}
对于await方法的执行主要可以分为6步：
- 把当前节点（线程）链接到条件队列
- 将当前节点（线程）持有的资源释放掉，并唤醒其后继节点（如有）
- 自旋判断当前节点（线程）是否在等待队列
- 如果是，则结束条件等待，执行第
4步（如发生中断也会结束条件等待） - 如果不是，则继续执行条件等待，执行第
3步（如被唤醒）
- 如果是，则结束条件等待，执行第
- 在等待队列中对于当前节点（线程）执行节点出队操作
- 如果出队成功，则返回并执行第
5步（证明资源获取成功） - 如果出队失败，则阻塞等待（自旋+阻塞，详情可阅读上文），并执行第
4步（如被唤醒）
- 如果出队成功，则返回并执行第
- 将条件队列中的取消节点移除（执行清理操作）
- 对等待过程中发生的中断进行处理（重新标记中断/抛出中断异常）
在基于
ConditionObject的条件队列上只有以独占的方式持有资源才能执行await操作（即将节点从等待队列转移到条件队列），所以我们在await方法中并没有看到很多类似于CAS的操作。
虽然说在await方法中将整个执行流程划分为6个步骤，但其中条件阻塞的核心流程就只有3步，分别是“将当前节点链接到条件队列”、“将当前节点从等待队列中移除”和“在条件队列中对当前节点执行阻塞操作”。
-
把当前节点（线程）链接到条件队列
/** * Adds a new waiter to wait queue. * @return its new wait node */ private Node addConditionWaiter() { Node t = lastWaiter; // If lastWaiter is cancelled, clean out. if (t != null && t.waitStatus != Node.CONDITION) { unlinkCancelledWaiters(); t = lastWaiter; } // 将节点构造且加入到条件队列 Node node = new Node(Thread.currentThread(), Node.CONDITION); if (t == null) firstWaiter = node; else t.nextWaiter = node; lastWaiter = node; return node; }在
addConditionWaiter方法中，首先会获取正常的队尾节点（如队尾节点被取消，则进行清理操作），然后再将当前线程构造为Node节点链接到条件队列的队尾（此处将节点状态设置为CONDITION状态，表示处于条件队列中）。此处需要注意的是，因为条件队列的操作都是在以独占方式持有资源的情况下进行的，所以队列中的节点并不存在临界状态。也就是说在条件队列中节点的
waitStatus只有两种情况，一是CONDITION状态，表示在条件队列呆着；二是CANCELLED状态，表示被取消了。因此在addConditionWaiter方法中遇到非CONDITION状态的节点直接当它是取消节点进行清理。其中，当发现队尾节点是
CANCELLED状态时（即节点被取消），则执行unlinkCancelledWaiters方法对取消节点进行清理：/** * Unlinks cancelled waiter nodes from condition queue. * Called only while holding lock. This is called when * cancellation occurred during condition wait, and upon * insertion of a new waiter when lastWaiter is seen to have * been cancelled. This method is needed to avoid garbage * retention in the absence of signals. So even though it may * require a full traversal, it comes into play only when * timeouts or cancellations occur in the absence of * signals. It traverses all nodes rather than stopping at a * particular target to unlink all pointers to garbage nodes * without requiring many re-traversals during cancellation * storms. */ private void unlinkCancelledWaiters() { Node t = firstWaiter; // trail用于记录被取消节点的前驱节点 Node trail = null; while (t != null) { // 获取当前节点的后继节点 Node next = t.nextWaiter; // 如果当前节点被取消了 if (t.waitStatus != Node.CONDITION) { // 清除与后继节点的链接 t.nextWaiter = null; // 如果当前节点是头节点，直接用firstWaiter指向被取消节点的后继节点（相当于丢掉当前取消节点） if (trail == null) firstWaiter = next; // 否则将前驱节点指向后继节点（相当于丢掉当前取消节点） else trail.nextWaiter = next; // 如果后继节点为null，直接用lastWaiter指向前驱节点（相当于丢掉当前取消节点） if (next == null) lastWaiter = trail; } else // 记录被取消节点的前驱节点 trail = t; // 继续遍历下一个节点 t = next; } }关于
unlinkCancelledWaiters方法的大致思路与从单向链表中删除节点并无差异，即从头节点开始逐个遍历至尾节点，期间遇到被取消的节点就进行清理操作（移除）。具体笔者已经在代码关键位置标注了注释，在这里就不再展开了。 -
将当前节点（线程）持有的资源释放掉，并唤醒其后继节点
在将线程构造为
Node节点链接到条件队列后，接着就是对等待队列中持有资源的节点执行释放操作。注意，在独占模式下持有资源的节点是（等待队列的）head节点。/** * Invokes release with current state value; returns saved state. * Cancels node and throws exception on failure. * @param node the condition node for this wait * @return previous sync state */ final int fullyRelease(Node node) { boolean failed = true; try { // 获取state值 int savedState = getState(); // 释放state if (release(savedState)) { failed = false; // 成功则返回，外层保存在起来，之后唤醒获取时会重新传入state return savedState; } else { throw new IllegalMonitorStateException(); } } finally { if (failed) node.waitStatus = Node.CANCELLED; } }在
fullyRelease方法中会通过调用release方法释放资源，并唤醒后继节点将当前head节点踢出队列。（具体release方法的执行逻辑可阅读上文）对于传统
Condition的定义，在通过条件队列进行阻塞后是会释放持有的资源（以让其他线程可以获取资源）。当然这取决于我们最终的实现，如果我们在实现中将这部分特性去掉，则必须在相应的使用文档上特别标明。 -
将当前节点（线程）陷入阻塞，直到被唤醒
在将节点从等待队列转移到条件队列后，接着就是将当前节点（线程）陷入阻塞状态了。这一步主要是基于
isOnSyncQueue方法的判断逻辑来实现的，即在不符合此条件的情况下会执行LockSupport#park使当前节点（线程）进入阻塞状态。/** * Returns true if a node, always one that was initially placed on * a condition queue, is now waiting to reacquire on sync queue. * @param node the node * @return true if is reacquiring */ final boolean isOnSyncQueue(Node node) { // 1. 判断当前节点状态是否为CONDITION，如果是则证明存在于条件队列，返回false if (node.waitStatus == Node.CONDITION) return false; // 2. 判断当前节点的前驱节点（prev链接）是否为null，如果是则证明并不存在于等待队列，返回false（waitStatus!=CONDITION） if (node.prev == null) return false; // 3. 判断当前节点的后继节点是否为null，如果是则证明存在于等待队列，返回true（waitStatus!=CONDITION && node.prev!=null） if (node.next != null) // If has successor, it must be on queue return true; /* * node.prev can be non-null, but not yet on queue because * the CAS to place it on queue can fail. So we have to * traverse from tail to make sure it actually made it. It * will always be near the tail in calls to this method, and * unless the CAS failed (which is unlikely), it will be * there, so we hardly ever traverse much. */ // 4. 判断当前节点是否存在于等待队列（通过从尾节点向前遍历），如果是则返回true，否则返回false return findNodeFromTail(node); } /** * Returns true if node is on sync queue by searching backwards from tail. * Called only when needed by isOnSyncQueue. * @return true if present */ private boolean findNodeFromTail(Node node) { Node t = tail; for (;;) { if (t == node) return true; if (t == null) return false; t = t.prev; } }在
isOnSyncQueue方法中会通过各种边界条件判断当前节点是否存在于等待队列，其中判断条件可分为4点：- 判断当前节点状态是否为
CONDITION，如果是则证明存在于条件队列，返回false因为条件队列中节点的
waitStatus都为CONDITION，所以此条件可以证明节点不存在于等待队列 - 判断当前节点的前驱节点（
prev链接）是否为null，如果是则证明并不存在于等待队列，返回false因为条件等待的线程会存在虚假唤醒的情况，如果此时刚好执行
signal方法（唤醒），并执行到如下临界代码处：boolean transferForSignal(Node node) { if (!compareAndSetWaitStatus(node, Node.CONDITION, 0)) return false; // o <== 执行临界代码处 Node p = enq(node); int ws = p.waitStatus; if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)) LockSupport.unpark(node.thread); return true; }不难看出此时在
waitStatus不等于CONDITION的情况下节点还是处于条件队列。而对于这种情况，AQS就通过对当前节点的前驱节点（prev链接）是否为null来判断其是否处于等待队列。（在等待队列中除了head节点外其他节点都存在前驱节点，即使等待队列为空的情况它也会创建一个虚拟head节点来作为其前驱节点，具体可阅读上文相关部分） - 判断当前节点的后继节点是否为
null，如果是则证明存在于等待队列，返回true因为对于等待队列的插入（如通过
enq方法）是先对prev链接进行设置，然后在CAS成功后（保证了插入原子性）再对next链接进行设置。即，在prev链接不为null时并不代表节点就已经插入到等待队列（存在CAS失败的情况），而在next链接也不为null时才可以证明节点已经插入到等待队列中，具体可阅读上文相关部分 - 判断当前节点是否存在于等待队列（通过从尾节点向前遍历），如果是则返回
true，否则返回false如果上述条件都不满足则无法判断节点是否在等待队列，此时需要从等待队列的尾节点逐个向前遍历判断当前节点是否存在于队列中
- 判断当前节点状态是否为
signal唤醒
在持有Lock的情况下，执行signal方法将会使（条件）等待时间最长的节点被唤醒，而执行signalAll方法则会将所有处于条件等待的节点唤醒。
使用
ConditionObject的前提条件是需要持有Lock，也就说只有在持有Lock的情况下才能调用signal方法或signalAll方法。
public class ConditionObject implements Condition, java.io Serializable {
/**
* Moves the longest-waiting thread, if one exists, from the
* wait queue for this condition to the wait queue for the
* owning lock.
*
* @throws IllegalMonitorStateException if {@link #isHeldExclusively}
* returns {@code false}
*/
public final void signal() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
// 获取第一个节点
Node first = firstWaiter;
if (first != null)
doSignal(first);
}
/**
* Moves all threads from the wait queue for this condition to
* the wait queue for the owning lock.
*
* @throws IllegalMonitorStateException if {@link #isHeldExclusively}
* returns {@code false}
*/
public final void signalAll() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
// 获取第一个节点
Node first = firstWaiter;
if (first != null)
doSignalAll(first);
}
}
此处通过
isHeldExclusively方法来判断当前线程是否独占地持有资源，如并没有持有则抛出异常IllegalMonitorStateException。
在signal和signalAll方法中，它们会将具体唤醒逻辑委托给了doSignal方法和doSignalAll方法（在独占式持有资源并且在条件队列中存在节点的情况下）。
/**
* Removes and transfers nodes until hit non-cancelled one or
* null. Split out from signal in part to encourage compilers
* to inline the case of no waiters.
* @param first (non-null) the first node on condition queue
*/
private void doSignal(Node first) {
// 将头节点开始向后逐个遍历
do {
// 将firstWaiter链接指向其后继节点
if ( (firstWaiter = first.nextWaiter) == null)
// 如果条件队列只有一个节点，则将lastWaiter链接也设置为null（firstWaiter链接已经被设置为null）
lastWaiter = null;
// 将当前节点的nextWaiter链接设置为null，表示将其踢出条件队列
first.nextWaiter = null;
} while (!transferForSignal(first) // 唤醒条件等待的节点，并判断操作是否成功，如果成功则直接退出循环（唤醒等待时间最长的节点），否则继续执行循环
&& (first = firstWaiter) != null); // 将first指向后一个节点，让其继续执行循环（直到队尾）
}
/**
* Removes and transfers all nodes.
* @param first (non-null) the first node on condition queue
*/
private void doSignalAll(Node first) {
// 将firstWaiter链接和lastWaiter链接设置为null
lastWaiter = firstWaiter = null;
// 将头节点开始向后逐个遍历
do {
// 获取当前节点的后继节点
Node next = first.nextWaiter;
// 将当前节点的nextWaiter链接设置为null，表示将其踢出条件队列
first.nextWaiter = null;
// 唤醒条件等待的节点
transferForSignal(first);
// 将first指向后一个节点，让其继续执行循环（直到队尾）
first = next;
} while (first != null);
}
而在doSignal和doSignalAll方法中，它们则会从传入first节点（条件等待队列中的头节点）开始向后逐个遍历并通过调用transferForSignal方法唤醒等待中的节点，其中对于唤醒等待时间最长的节点还是唤醒所有处于条件等待的节点则取决于两者所定义的语义了。这里，我们再来看一下真正执行唤醒操作的transferForSignal方法是如何实现的：
/**
* Transfers a node from a condition queue onto sync queue.
* Returns true if successful.
* @param node the node
* @return true if successfully transferred (else the node was
* cancelled before signal)
*/
final boolean transferForSignal(Node node) {
/*
* If cannot change waitStatus, the node has been cancelled.
*/
// 先将节点状态从CONDITION转为0，如果节点被取消则此处会执行失败，即返回false
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节点插入到等待队列
Node p = enq(node);
int ws = p.waitStatus;
// 如果当前节点被取消或者其前驱节点状态变更失败，则将当前node节点（线程）唤醒
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
return true;
}
对于transferForSignal方法主要分为3个步骤：
- 将当前节点的状态从
CONDITION变更为0，如果失败直接返回false（表示当前节点被取消了） - 将当前节点通过
enq方法插入到等待队列（外层代码已经将节点从条件队列中出队，所以来到这里相对来说已经转移成功了） - 如果当前节点被取消或者其前驱节点状态变更失败（变更为
SIGNAL），则通过LockSupport#unpark方法将当前节点（线程）唤醒
对于等待队列的插入，我们是需要将其前驱节点设置为SIGNAL状态才算真正完成（以表示其存在后继节点需要被唤醒，具体可阅读上文相关部分）。因此在第3步中如果成功地将其前驱节点设置为SIGNAL状态则表示已经成功地将节点从条件队列转移到等待队列，而如果其前驱节点被取消或者SIGNAL状态设置失败则还需要通过LockSupport#unpark方法将当前节点（线程）唤醒并让它完成对异常地修正（跳过被取消节点或者设置前驱节点的状态）。最终回到await方法的第3步继续执行：
public final void await() throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
//...
// 3. 自旋判断当前节点（线程）是否在等待队列
while (!isOnSyncQueue(node)) {
LockSupport.park(this);
// ...
}
// 4. 在等待队列中对于当前节点（线程）执行节点出队操作
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
// ...
}
除了在调用signal和signalAll方法将节点转移回等待队列出现异常时会触发await方法让其继续执行第3步判断（唤醒）外，节点在正常从条件队列转移回等待队列后经过前驱节点（head节点）执行完毕唤醒其后继节点的情况也同样会触发await方法第3步继续执行（唤醒）。其中，两种同样会跳出第3步循环（因为已经转移回等待队列），并执行acquireQueued方法进行修正、等待和获取（具体逻辑可阅读上文相关部分）。
总的来说，对于
await方法的执行本质上是将节点从等待队列转移到条件队列，而对于signal方法的执行则本质是将节点从条件队列转移到等待队列，其中对于节点处于哪一个队列则是通过Node节点的waitStatus状态进行判断。
总结
本文由浅入深对AQS进行了解读，其中主要包括其数据结构、等待队列和条件队列等。然而虽然AQS从原理上看是十分精细并复杂，但是作为同步/阻塞的框架使用起来所需要做的事情其实并不多，下面笔者将从使用者的角度列举出AQS的用法。
语义实现
在AQS的实现中，我们的首要任务是对资源获取语义进行实现，具体所需要实现的方法如下表所示：
| 方法 | 描述 |
|---|---|
tryAcquire |
表示在排他模式(EXCLUSIVE)下去获取资源，如果返回true表示获取成功，否则表示获取失败。其中，在方法的实现中我们应该判断当前是否能在独占模式获取资源。 |
tryRelease |
表示在排他模式(EXCLUSIVE)下去释放资源，如果返回true表示全部释放成功，否则表示释放失败或者部分释放。 |
tryAcquireShared |
表示在共享模式(SHARED)下去去获取资源，如果返回大于0表示获取成功并且其后继节点也可能成功获取资源；如果返回等于0表示获取成功但其后继节点不能再成功获取资源了；如果返回小于0则表示获取失败。其中，在方法的实现中我们应该判断当前是否能够在共享模式下获取资源。 |
tryReleaseShared |
表示在共享模式(SHARED)下去释放资源，如果返回true表示释放成功，否则表示释放失败。 |
isHeldExclusively |
表示资源是否被独占地持有，如果返回true表示被独占持有，否则表示没有被独占持有。 |
对于
AQS条件等待机制的使用，我们是需要实现isHeldExclusively方法的，因为会存在类似于signal和signalAll这样的方法运用到它。而如果我们并没有运用到Condition，在实现类中是可以不实现isHeldExclusively方法的（如其他地方也没有运用到）。
以上就是我们在使用AQS时所需要完成的所有工作了，至于等待队列或条件队列相关的实现细节我们是无需关心的，因为这是AQS自身的工作。
方法使用
而在我们对上述语义进行实现的过程中，或者在使用AQS时需要对其进行一定的监控时，我们可以使用AQS提供给我们的一些方法来更快速地完成工作，下面笔者将相关部分分点列出：
-
用于操作
state的方法方法 描述 getState获取 state的值setState设置 state的值compareAndSetState通过 CAS修改state的值这些方法都是
protected和final的，主要（只能）用于AQS的实现类中，并且不能被继承修改。一般用于实现获取/释放资源的语义。 -
用于操作执行线程的方法
方法 描述 getExclusiveOwnerThread获取当前执行线程 setExclusiveOwnerThread设置当前执行线程 这些方法主要用于实现排他模式(
EXCLUSIVE)，并且可以通过比较当前线程是否与保存中的线程相同来实现可重入。 -
用于获取/释放资源的方法
方法 描述 tryAcquire表示在排他模式( EXCLUSIVE)下去获取资源，如果返回true表示获取成功，否则表示获取失败。tryRelease表示在排他模式( EXCLUSIVE)下去释放资源，如果返回true表示全部释放成功，否则表示释放失败或者部分释放。tryAcquireShared表示在共享模式( SHARED)下去去获取资源，如果返回大于0表示获取成功并且其后继节点也可能成功获取资源；如果返回等于0表示获取成功但其后继节点不能再成功获取资源了；如果返回小于0则表示获取失败。tryReleaseShared表示在共享模式( SHARED)下去释放资源，如果返回true表示释放成功，否则表示释放失败。isHeldExclusively表示资源是否被独占地持有，如果返回 true表示被独占持有，否则表示没有被独占持有。acquire表示在排他模式( EXCLUSIVE)下去获取资源，如果获取失败会陷入阻塞（进入等待队列）直到获取成功。acquireInterruptibly表示在排他模式( EXCLUSIVE)下去获取资源，获取失败会陷入阻塞（进入等待队列）直到获取成功或中断抛出异常。tryAcquireNanos表示在排他模式( EXCLUSIVE)下在规定时间内去获取资源，获取失败会陷入阻塞（进入等待队列）直到获取成功或中断抛出异常，其中如果在规定时间内获取成功会返回true，超时则返回false。release表示在排他模式( EXCLUSIVE)下去释放资源，如果释放成功返回true，否则返回false。acquireShared表示在共享模式( SHARED)下去获取资源，获取失败会陷入阻塞（进入等待队列）直到获取成功（与排他模式相比，此方法可以让多个线程同时获取到资源）。acquireSharedInterruptibly表示在共享模式( SHARED)下去获取资源，获取失败会陷入阻塞（进入等待队列）直到获取成功或中断抛出异常（与排他模式相比，此方法可以让多个线程同时获取到资源）。tryAcquireSharedNanos表示在共享模式( SHARED)下去获取资源，获取失败会陷入阻塞（进入等待队列）直到获取成功或中断抛出异常，其中如果在规定时间内获取成功会返回true，超时则返回false。（与排他模式相比，此方法可以让多个线程同时获取到资源）。releaseShared表示在共享模式( SHARED)下去释放资源，如果释放成功返回true，否则返回false。这些方法都是
public的，主要用于对AQS实现类（如ReentrantLock）的使用，如获取/释放资源的操作。 -
用于查看等待队列的方法
方法 描述 hasQueuedThreads查看等待队列当前是否存在等待节点（线程）。需要注意的是由于 head节点并非正在等待的节点，所以并不算在内。另外由于中断和超时导致节点被取消在任何时候都有可能发生，所以这里并不保证值是正确的。hasContended查看等待队列是否曾经存在过等待节点（线程），即是否曾经发生过竞争。 getFirstQueuedThread查看等待队列中目前正在等待且等待时间最长的节点（线程），如果不存在返回 null。需要注意的是由于head节点并非正在等待的节点，所以并不算在内。isQueue查看当前入参节点（线程）是否存在等待队列中。 hasQueuedPredecessors查看等待队列中当前是否存在比当前节点（线程）等待时间长的等待节点（线程），即是否存在等待节点（线程）位于当前节点（线程）的前面。需要注意的是由于 head节点并非正在等待的节点，所以并不算在内。另外由于中断和超时导致节点被取消在任何时候都有可能发生，所以这里并不保证值是正确的。这些方法都是
public的，主要用于在AQS实现类中判断是否能获取/释放资源（具体需看语义），或者用于对AQS条件队列的监控。另外，
AQS中还存在一个非public的方法apparentlyFirstQueuedIsExclusive（默认修饰符），用于查看等待队列中第一个正在等待的节点（线程）是否为排他模式，此方法目前仅仅应用于ReentrantReadWriteLock。其中，使用默认修饰符的原因也是因为它仅被ReentrantReadWriteLock使用（ReentrantReadWriteLock与AQS位于同一个包下）。 -
用于监控等待队列的方法
方法 描述 getQueueLength用于获取等待队列的长度（预估值）。 getQueuedThreads用于获取等待队列中的节点（线程）列表（预估值）。 getExclusiveQueuedThreads用于获取等待队列中处于排他模式( EXCLUSIVE)的节点（线程）列表（预估值）。getSharedQueuedThreads用于获取等待队列中处于共享模式( SHARED)的节点（线程）列表（预估值）。这些方法是
public的，一般用于对AQS等待队列使用情况的监控。其中，方法的返回值都是一个预估值，这是因为等待队列是基于链表实现的，对于上述方法的结果都需要逐个遍历，而在遍历的过程中节点是可以动态改变的，所以最终得到的只是一个预估值。 -
用于监控条件队列的方法
方法 描述 owns用于判断传入的 Condition（条件队列）是否用于当前AQS，即是否通过当前AQS所创建的。hasWaiters用于判断传入 Condition（条件队列）中是否存在正在等待的节点（线程）（预估值），另外如果传入Condition（条件队列）不属于当前AQS则抛出异常。getWaitQueueLength用于查看传入 Condition（条件队列）中当前等待节点（线程）的数量（预估值），另外如果传入Condition（条件队列）不属于当前AQS则抛出异常。getWaitingThreads用于查看传入 Condition（条件队列）中当前等待节点（线程）的集合（预估值），另外如果传入Condition（条件队列）不属于当前AQS则抛出异常。这些方法是
public的，一般用于对AQS条件队列使用情况的监控。其中，方法的返回值都是一个预估值，这是因为条件队列是基于链表实现的，对于上述方法的结果都需要逐个遍历，而在遍历的过程中节点是可以动态改变的，所以最终得到的只是一个预估值。
实现例子
最终，在这里笔者贴出了官方文档给出的AQS实现例子，我们可以结合上下文来阅读以加深对AQS的理解。
/*
* <p>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:
*/
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));
}
}
/*
* <p>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.
*/
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);
}
}
在例子中分别实现了排他模式(EXCLUSIVE)和共享模式(SHARED)，其中具体的实现细节可在上文相应的部分找到答案，在这里就不再细讲了。但也许你会注意到，在例子中AQS的实现类都是担任着内部类，然后再委托给外部类去调用，这一点官方文档也有所提及，这里笔者把相关注释说明贴出来：
/*
* <p>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.
*/
即，因为在AQS中并没有实现任何锁/同步器的接口（并不符合使用规范），所以我们在使用过程中应该通过内部类委托的方式来实现。
接口具有封装性的作用，对于大部分调用者来说他们只认相关的锁/同步器的接口，如果直接使用
AQS的话会对调用者很不友好。
评论区