Java多线程之并发编程的核心AQS详解 Java多线程之并发编程的核心

一、AQS简介

1.1、AOS概念
1.2、AQS的核心思想
1.3、AQS是自旋锁
1.4、AQS支持两种资源分享的方式

二、AQS原理

2.1、同步状态的管理
2.2、等待队列
2.3、CLH队列中的结点
2.4、队列定义
2.5、AQS底层的CAS机制
2.6、通过ReentrantLock理解AQS

三、AQS方法

3.1、用户需要自己重写的方法
3.2、AQS 提供的一系列模板方法
3.3、acquire(int)方法
3.4、release(int)方法
3.5、acquireShared(int)方法
3.6、releaseShared(int)方法
参考链接：

总结

前言：Java并发包很多的同步工具类底层都是基于AQS来实现的，比如我们工作中经常用的Lock工具ReentrantLock、栅栏CountDownLatch、信号量Semaphore等。如果你想深入研究Java并发编程的话，那么AQS一定是绕不开的一块知识点，而且关于AQS的知识点也是面试中经常考察的内容，所以深入学习AQS很有必要。
学习AQS之前，我们有必要了解一下AQS底层中大量使用的CAS：Java多线程10：并发编程的基石CAS机制

一、AQS简介
1.1、AOS概念

AQS，全名AbstractQueuedSynchronizer，是一个抽象类的队列式同步器，它的内部通过维护一个状态volatile int state(共享资源)，一个FIFO线程等待队列来实现同步功能。AQS类是整个 JUC包的核心类，JUC 中的ReentrantLock、ReentrantReadWriteLock、CountDownLatch、Semaphore和LimitLatch等同步工具都是基于AQS实现的。

state用关键字volatile修饰，代表着该共享资源的状态一更改就能被所有线程可见，而AQS的加锁方式本质上就是多个线程在竞争state，当state为0时代表线程可以竞争锁，不为0时代表当前对象锁已经被占有，其他线程来加锁时则会失败，加锁失败的线程会被放入一个FIFO的等待队列中，这些线程会被UNSAFE.park()操作挂起，等待其他获取锁的线程释放锁才能够被唤醒。
而这个等待队列其实就相当于一个CLH队列，用一张原理图来表示大致如下：

文章图片

1.2、AQS的核心思想

如果被请求的共享资源空闲，则将当前请求资源的线程设置为有效的工作线程，并将共享资源设置为锁定状态，如果被请求的共享资源被占用，那么就需要一套线程阻塞等待以及被唤醒时锁分配的机制，这个机制AQS是用CLH队列锁实现的，即将暂时获取不到锁的线程加入到队列中。CLH（Craig，Landin，and Hagersten）队列是一个虚拟的双向队列，虚拟的双向队列即不存在队列实例，仅存在节点之间的关联关系。
AQS是将每一条请求共享资源的线程封装成一个CLH锁队列的一个结点（Node），来实现锁的分配。

用大白话来说，AQS就是基于CLH队列，用volatile修饰共享变量state，线程通过CAS去改变状态符，成功则获取锁成功，失败则进入等待队列，等待被唤醒。

文章图片

1.3、AQS是自旋锁
AQS是自旋锁：在等待唤醒的时候，经常会使用自旋（while(!cas())）的方式，不停地尝试获取锁，直到被其他线程获取成功
实现了AQS的锁有：自旋锁、互斥锁、读锁写锁、条件产量、信号量、栅栏都是AQS的衍生物

1.4、AQS支持两种资源分享的方式
Exclusive（独占，只有一个线程能执行，如ReentrantLock）和Share（共享，多个线程可同时执行，如Semaphore/CountDownLatch）。
自定义的同步器继承AQS后，只需要实现共享资源state的获取和释放方式即可，其他如线程队列的维护（如获取资源失败入队/唤醒出队等）等操作，AQS在底层已经实现了。
线程的阻塞和唤醒
在JDK1.5之前，除了内置的监视器机制外，没有其它方法可以安全且便捷得阻塞和唤醒当前线程。
JDK1.5以后，java.util.concurrent.locks包提供了LockSupport类来作为线程阻塞和唤醒的工具。

二、AQS原理
2.1、同步状态的管理
同步状态，其实就是资源。AQS使用单个int（32位）来保存同步状态，并暴露出getState、setState以及compareAndSetState操作来读取和更新这个状态。

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizerimplements 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 thisreturn unsafe.compareAndSwapInt(this, stateOffset, expect, update); } //省略展示其它代码...}

这几个方法都是Final修饰的，说明子类中无法重写它们。我们可以通过修改State字段表示的同步状态来实现多线程的独占模式和共享模式（加锁过程）。

文章图片

文章图片

2.2、等待队列
等待队列，是AQS框架的核心，整个框架的关键其实就是如何在并发状态下管理被阻塞的线程。
等待队列是严格的FIFO队列，是Craig，Landin和Hagersten锁（CLH锁）的一种变种，采用双向循环链表实现，因此也叫CLH队列。

2.3、CLH队列中的结点
AQS内部还定义了一个静态类Node，表示CLH队列的每一个结点，该结点的作用是对每一个等待获取资源做了封装，包含了需要同步的线程本身、线程等待状态....
LH队列中的结点是对线程的包装，结点一共有两种类型：独占（EXCLUSIVE）和共享（SHARED）。
每种类型的结点都有一些状态，其中独占结点使用其中的CANCELLED(1)、SIGNAL(-1)、CONDITION(-2)，共享结点使用其中的CANCELLED(1)、SIGNAL(-1)、PROPAGATE(-3)。

结点状态	值	描述
CANCELLED	1	取消。表示后驱结点被中断或超时，需要移出队列
SIGNAL	-1	发信号。表示后驱结点被阻塞了（当前结点在入队后、阻塞前，应确保将其prev结点类型改为SIGNAL，以便prev结点取消或释放时将当前结点唤醒。）
CONDITION	-2	Condition专用。表示当前结点在Condition队列中，因为等待某个条件而被阻塞了
PROPAGATE	-3	传播。适用于共享模式（比如连续的读操作结点可以依次进入临界区，设为PROPAGATE有助于实现这种迭代操作。）
INITIAL	0	默认。新结点会处于这种状态

2.4、队列定义
对于CLH队列，当线程请求资源时，如果请求不到，会将线程包装成结点，将其挂载在队列尾部。
下面结合代码一起看下节点进入队列的过程。

private Node enq(final Node node) {for (; ; ) {Node t = tail; // 1if (t == null) { // Must initializeif (compareAndSetHead(new Node()))// 2 tail = head; } else {node.prev = t; // 3if (compareAndSetTail(t, node)) { // 4t.next = node; return t; }}}}

2.5、AQS底层的CAS机制
在研究JDK中AQS时，会发现这个类很多地方都使用了CAS操作，在并发实现中CAS操作必须具备原子性，而且是硬件级别的原子性，Java被隔离在硬件之上，明显力不从心，这时为了能直接操作操作系统层面，肯定要通过用C++编写的native本地方法来扩展实现。JDK提供了一个类来满足CAS的要求，sun.misc.Unsafe，从名字上可以大概知道它用于执行低级别、不安全的操作，AQS就是使用此类完成硬件级别的原子操作。UnSafe通过JNI调用本地C++代码，C++代码调用CPU硬件指令集。
Unsafe是一个很强大的类，它可以分配内存、释放内存、可以定位对象某字段的位置、可以修改对象的字段值、可以使线程挂起、使线程恢复、可进行硬件级别原子的CAS操作等等。

2.6、通过ReentrantLock理解AQS
ReentrantLock中公平锁和非公平锁在底层是相同的，这里以非公平锁为例进行分析。
在非公平锁中，有一段这样的代码：

// java.util.concurrent.locks.ReentrantLock static final class NonfairSync extends Sync {...final void lock() {if (compareAndSetState(0, 1))setExclusiveOwnerThread(Thread.currentThread()); elseacquire(1); }...}

看一下这个Acquire是怎么写的：

// java.util.concurrent.locks.AbstractQueuedSynchronizer public final void acquire(int arg) {if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))selfInterrupt(); }

再看一下tryAcquire方法：

// java.util.concurrent.locks.AbstractQueuedSynchronizer protected boolean tryAcquire(int arg) {throw new UnsupportedOperationException(); }

可以看出，这里只是AQS的简单实现，具体获取锁的实现方法是由各自的公平锁和非公平锁单独实现的（以ReentrantLock为例）。如果该方法返回了True，则说明当前线程获取锁成功，就不用往后执行了；如果获取失败，就需要加入到等待队列中。

三、AQS方法 AQS代码内部提供了一系列操作锁和线程队列的方法，主要操作锁的方法包含以下几个：
compareAndSetState()：利用CAS的操作来设置state的值
tryAcquire(int)：独占方式获取锁。成功则返回true，失败则返回false。
tryRelease(int)：独占方式释放锁。成功则返回true，失败则返回false。
tryReleaseShared(int)：共享方式释放锁。如果释放后允许唤醒后续等待结点返回true，否则返回false。
像ReentrantLock就是实现了自定义的tryAcquire-tryRelease，从而操作state的值来实现同步效果。

3.1、用户需要自己重写的方法
上面介绍到 AQS 已经帮用户解决了同步器定义过程中的大部分问题，只将下面两个问题丢给用户解决：

什么是资源
什么情况下资源是可以被访问的

具体的，AQS 是通过暴露以下 API 来让用户解决上面的问题的。

钩子方法	描述
tryAcquire	独占方式。尝试获取资源，成功则返回true，失败则返回false。
tryRelease	独占方式。尝试释放资源，成功则返回true，失败则返回false。
tryAcquireShared	共享方式。尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。
tryReleaseShared	共享方式。尝试释放资源，如果释放后允许唤醒后续等待结点返回true，否则返回false。
isHeldExclusively	该线程是否正在独占资源。只有用到condition才需要去实现它。

如果你需要实现一个自己的同步器，一般情况下只要继承 AQS ，并重写 AQS 中的这个几个方法就行了。至于具体线程等待队列的维护（如获取资源失败入队/唤醒出队等），AQS已经在顶层实现好了。要不怎么说Doug Lea贴心呢。
需要注意的是：如果你没在子类中重写这几个方法就直接调用了，会直接抛出异常。所以，在你调用这些方法之前必须重写他们。不使用的话可以不重写。

3.2、AQS 提供的一系列模板方法
查看 AQS 的源码我们就可以发现这个类提供了很多方法，看起来让人“眼花缭乱”的。但是最主要的两类方法就是获取资源的方法和释放资源的方法。因此我们抓住主要矛盾就行了：

public final void acquire(int arg) // 独占模式的获取资源
public final boolean release(int arg) // 独占模式的释放资源
public final void acquireShared(int arg) // 共享模式的获取资源
public final boolean releaseShared(int arg) // 共享模式的释放资源

3.3、acquire(int)方法
该方法以独占方式获取资源，如果获取到资源，线程继续往下执行，否则进入等待队列，直到获取到资源为止，且整个过程忽略中断的影响。该方法是独占模式下线程获取共享资源的顶层入口。

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

下面分析下这个acquire方法的具体执行流程：

step1：首先这个方法调用了用户自己实现的方法tryAcquire方法尝试获取资源，如果这个方法返回true，也就是表示获取资源成功，那么整个acquire方法就执行结束了，线程继续往下执行；
step2：如果tryAcquir方法返回false，也就表示尝试获取资源失败。这时acquire方法会先调用addWaiter方法将当前线程封装成Node类并加入一个FIFO的双向队列的尾部。
step3：再看acquireQueued这个关键方法。首先要注意的是这个方法中哪个无条件的for循环，这个for循环说明acquireQueued方法一直在自旋尝试获取资源。进入for循环后，首先判断了当前节点的前继节点是不是头节点，如果是的话就再次尝试获取资源，获取资源成功的话就直接返回false（表示未被中断过）
假如还是没有获取资源成功，判断是否需要让当前节点进入waiting状态，经过 shouldParkAfterFailedAcquire这个方法判断，如果需要让线程进入waiting状态的话，就调用LockSupport的park方法让线程进入waiting状态。进入waiting状态后，这线程等待被interupt或者unpark（在release操作中会进行这样的操作，可以参见后面的代码）。这个线程被唤醒后继续执行for循环来尝试获取资源。

final boolean acquireQueued(final Node node, int arg) {boolean failed = true; try {boolean interrupted = false; for (; ; ) {final Node p = node.predecessor(); //首先判断了当前节点的前继节点是不是头节点，如果是的话就再次尝试获取资源，//获取资源成功的话就直接返回false（表示未被中断过）if (p == head && tryAcquire(arg)) {setHead(node); p.next = null; // help GCfailed = false; return interrupted; }//判断是否需要让当前节点进入waiting状态if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())// 如果在整个等待过程中被中断过，则返回true，否则返回false。// 如果线程在等待过程中被中断过，它是不响应的。只是获取资源后才再进行自我中断selfInterrupt()，将中断补上。interrupted = true; }} finally {if (failed)cancelAcquire(node); }}

以上就是acquire方法的简单分析。
单独看这个方法的话可能会不太清晰，结合ReentrantLock、ReentrantReadWriteLock、CountDownLatch、Semaphore和LimitLatch等同步工具看这个代码的话就会好理解很多。

3.4、release(int)方法
release(int)方法是独占模式下线程释放共享资源的顶层入口。它会释放指定量的资源，如果彻底释放了（即state=0）,它会唤醒等待队列里的其他线程来获取资源。

public final boolean release(int arg) {if (tryRelease(arg)) {Node h = head; if (h != null && h.waitStatus != 0)unparkSuccessor(h); return true; }return false; } //上面已经讲过了，需要用户自定义实现protected boolean tryRelease(int arg) {throw new UnsupportedOperationException(); } 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); }

与acquire()方法中的tryAcquire()类似，tryRelease()方法也是需要独占模式的自定义同步器去实现的。正常来说，tryRelease()都会成功的，因为这是独占模式，该线程来释放资源，那么它肯定已经拿到独占资源了，直接减掉相应量的资源即可(state-=arg)，也不需要考虑线程安全的问题。
但要注意它的返回值，上面已经提到了，release()是根据tryRelease()的返回值来判断该线程是否已经完成释放掉资源了！所以自义定同步器在实现时，如果已经彻底释放资源(state=0)，要返回true，否则返回false。
unparkSuccessor(Node)方法用于唤醒等待队列中下一个线程。这里要注意的是，下一个线程并不一定是当前节点的next节点，而是下一个可以用来唤醒的线程，如果这个节点存在，调用unpark()方法唤醒。
总之，release()是独占模式下线程释放共享资源的顶层入口。它会释放指定量的资源，如果彻底释放了（即state=0）,它会唤醒等待队列里的其他线程来获取资源。（需要注意的是队列中被唤醒的线程不一定能立马获取资源，因为资源在释放后可能立马被其他线程（不是在队列中等待的线程）抢掉了）

3.5、acquireShared(int)方法
acquireShared(int)方法是共享模式下线程获取共享资源的顶层入口。它会获取指定量的资源，获取成功则直接返回，获取失败则进入等待队列，直到获取到资源为止，整个过程忽略中断。

public final void acquireShared(int arg) {//tryAcquireShared需要用户自定义实现if (tryAcquireShared(arg) < 0)doAcquireShared(arg); }

可以发现，这个方法的关键实现其实是获取资源失败后，怎么管理线程。也就是doAcquireShared的逻辑。

//不响应中断private void doAcquireShared(int arg) {final Node node = addWaiter(Node.SHARED); boolean failed = true; try {boolean interrupted = false; for (; ; ) {final Node p = node.predecessor(); if (p == head) {int r = tryAcquireShared(arg); if (r >= 0) {setHeadAndPropagate(node, r); p.next = null; // help GCif (interrupted)selfInterrupt(); failed = false; return; }}if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())interrupted = true; }} finally {if (failed)cancelAcquire(node); }}

可以看出，doAcquireShared的逻辑和acquireQueued的逻辑差不多。将当前线程加入等待队列尾部休息，直到其他线程释放资源唤醒自己，自己成功拿到相应量的资源后才返回。
简单总结下acquireShared的流程：

step1：tryAcquireShared()尝试获取资源，成功则直接返回；
step2：失败则通过doAcquireShared()进入等待队列park()，直到被unpark()/interrupt()并成功获取到资源才返回。整个等待过程也是忽略中断的。

3.6、releaseShared(int)方法
releaseShared(int)方法是共享模式下线程释放共享资源的顶层入口。它会释放指定量的资源，如果成功释放且允许唤醒等待线程，它会唤醒等待队列里的其他线程来获取资源。

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

释放掉资源后，唤醒后继。跟独占模式下的release()相似，但有一点稍微需要注意：独占模式下的tryRelease()在完全释放掉资源（state=0）后，才会返回true去唤醒其他线程，这主要是基于独占下可重入的考量；而共享模式下的releaseShared()则没有这种要求，共享模式实质就是控制一定量的线程并发执行，那么拥有资源的线程在释放掉部分资源时就可以唤醒后继等待结点。

参考链接：
从ReentrantLock的实现看AQS的原理及应用
并发编程的基石——AQS类

总结 【Java多线程之并发编程的核心AQS详解】本篇文章就到这里了，希望能够给你带来帮助，也希望您能够多多关注脚本之家的更多内容！