Java并发JUC——AQS

为什么需要AQS

• 锁和协作类有共同点：闸门
  • 像ReentrantLock和Semaphore有一些共同点，并且很相似
  • 事实上，不仅仅是ReentrantLock和Semaphore，包括CountDownLatch、ReentrantReadWriteLock都有这样类似的“协作”(或者叫同步)功能，其实它们底层都使用了同一个共同的基类——AQS
• 像上面提到的那些协作类，它们有很多工作都是类似的，所以如果能提取出一个工具类，那么就可以直接使用，对于ReentrantLock和Semaphore而言就可以屏蔽很多细节，只需要关注它们自己的“业务逻辑”就可以了

Semaphore和AQS的关系

• Semaphore内部有一个Sync类，Sync继承了AQS
• CountDownLatch也是一样的

AQS的重要性和地位

• AbstractQueuedSynchronizer是大名鼎鼎的Doug Lea写的，从JDK1.5加入的一个基于FIFO等待队列实现的一个用于实现同步器的基础框架，使用IDEA查看AQS的实现类，可以发现实现类如下：

AQS介绍

AQS：AbstractQuenedSynchronizer抽象的队列式同步器。是除了java自带的synchronized关键字之外的锁机制。 AQS的全称为（AbstractQueuedSynchronizer），这个类在java.util.concurrent.locks包

 

AQS是一个抽象类，主是是以继承的方式使用。AQS本身是没有实现任何同步接口的，它仅仅只是定义了同步状态的获取和释放的方法来供自定义的同步组件的使用。在java的同步组件中，AQS的子类（Sync等）一般是同步组件的静态内部类，即通过组合的方式使用。

 

抽象的队列式的同步器，AQS定义了一套多线程访问共享资源的同步器框架，许多同步类实现都依赖于它，如常用的ReentrantLock/Semaphore/CountDownLatch

 

它维护了一个volatile int state（代表共享资源）和一个FIFO（双向队列）线程等待队列（多线程争用资源被阻塞时会进入此队列）

 

AQS的核心思想是，如果被请求的共享资源空闲，则将当前请求资源的线程设置为有效的工作线程，并将共享资源设置为锁定状态，如果被请求的共享资源被占用，那么就需要一套线程阻塞等待以及被唤醒时锁分配的机制，这个机制AQS是用CLH队列锁实现的，即将暂时获取不到锁的线程加入到队列中。

 

CLH（Craig，Landin，and Hagersten）队列是一个虚拟的双向队列，虚拟的双向队列即不存在队列实例，仅存在节点之间的关联关系。 AQS是将每一条请求共享资源的线程封装成一个CLH锁队列的一个结点（Node），来实现锁的分配。

 

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

 

**注意：AQS是自旋锁：**在等待唤醒的时候，经常会使用自旋（while(!cas())）的方式，不停地尝试获取锁，直到被其他线程获取成功

实现了AQS的锁有：自旋锁、互斥锁、读锁写锁、条件产量、信号量、栅栏都是AQS的衍生物

AQS的具体实现方式如下：

AQS内部原理解析

AQS最核心的就是3大部分

• state
• 控制线程抢锁和配合的FIFO队列
• 期望协作工具类去实现的获取/释放等重要方法

state状态

AQS维护了一个private volatile int state;和一个FIFO线程等待队列，多线程争用资源被阻塞的时候就会进入这个队列。state就是共享资源，其访问方式有如下三种：getState()、setState()、compareAndSetState();

• 这里的state的具体含义，会根据具体实现类的不同而不同，比如在Semaphore里，它表示“剩余的许可证的数量”，而在CountDownLatch里，它表示“还需要到数的数量”
• state是volatile 修饰的，会被并发的修改，所以所有修改state的方法都需要保证线程安全，比如getState()、setState()、compareAndSetState()操作来读取和更新这个状态，这些方法都依赖与java.util.concurrent.atomic包的支持

在ReentrantLock中

• state用来表示“锁”的占有情况，包括可重入计数
• 当state的值为0的时候，表示该Lock不被任何线程所占有

控制线程抢锁和配合的FIFO队列

• 这个队列用来存放“等待的线程”，AQS就是“排队管理器”，当多个线程争用同一把锁时，必须有排队机制将那些没能拿到锁的线程串在一起，当锁释放时，锁管理器就会挑选一个合适的线程来占用这个刚刚释放的锁
• AQS会维护一个等待的线程队列，把线程都放到这个队列里。这个队列是一个虚拟的双向队列，虚拟的双向队列即不存在队列实例，仅存在节点之间的关联关系。

期望协作工具类去实现的获取/释放等重要方法

• 这里的获取和释放方法，是利用AQS的协作工具类里最重要的方法，是由协作类自己去实现的，并且含义各不相同

获取方法

• 获取操作会依赖state变量，经常会阻塞(比如获取不到锁的时候)
• 在Semaphore中，获取就是acquire()方法，作用是获取一个许可证
• 在CountDownLatch里面，获取就是await方法，作用是“等待，直到倒数结束”

释放方法

• 释放操作不会阻塞
• 在Semaphore中，释放方法就是release()方法，作用是释放一个许可证
• 在CountDownLatch里面，释放就是countDown()方法，作用就是“倒数1个数”

协作工具类需要重写tryAcquire()和tryRelease()等方法

AQS 定义了两种资源共享方式：

• 1、Exclusive：独占，只有一个线程能执行，如ReentrantLock
• 2、Share：共享，多个线程可以同时执行，如Semaphore、CountDownLatch、ReadWriteLock，CyclicBarrier
• 不同的自定义的同步器争用共享资源的方式也不同。

AQS底层使用了模板方法模式

同步器的设计是基于模板方法模式的，如果需要自定义同步器一般的方式是这样（模板方法模式很经典的一个应用）：

• 使用者继承AbstractQueuedSynchronizer并重写指定的方法。（这些重写方法很简单，无非是对于共享资源state的获取和释放）
• 将AQS组合在自定义同步组件的实现中，并调用其模板方法，而这些模板方法会调用使用者重写的方法。这和我们以往通过实现接口的方式有很大区别，这是模板方法模式很经典的一个运用。

自定义同步器在实现的时候只需要实现共享资源state的获取和释放方式即可，至于具体线程等待队列的维护，AQS已经在顶层实现好了。自定义同步器实现的时候主要实现下面几种方法：

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

ReentrantLock为例，（可重入独占式锁）：state初始化为0，表示未锁定状态，A线程lock()时，会调用tryAcquire()独占锁并将state+1.之后其他线程再想tryAcquire的时候就会失败，直到A线程unlock（）到state=0为止，其他线程才有机会获取该锁。A释放锁之前，自己也是可以重复获取此锁（state累加），这就是可重入的概念。 注意：获取多少次锁就要释放多少次锁，保证state是能回到零态的。

 

以CountDownLatch为例，任务分N个子线程去执行，state就初始化 为N，N个线程并行执行，每个线程执行完之后countDown（）一次，state就会CAS减一。当N子线程全部执行完毕，state=0，会unpark()主调用线程，主调用线程就会从await()函数返回，继续之后的动作。

 

一般来说，自定义同步器要么是独占方法，要么是共享方式，他们也只需实现tryAcquire-tryRelease、tryAcquireShared-tryReleaseShared中的一种即可。但AQS也支持自定义同步器同时实现独占和共享两种方式，如ReentrantReadWriteLock。 　在acquire() acquireShared()两种方式下，线程在等待队列中都是忽略中断的，acquireInterruptibly()/acquireSharedInterruptibly()是支持响应中断的。

AQS用法

• 1、写一个类，想好协作的逻辑，实现获取/释放方法
• 2、内部写一个Sync类继承AbstractQueuedSynchronizer
• 3、根据是否独占来重写tryAcquire/tryRelease或tryAcquireShared(int acquires)和tryReleaseShared(int releases)等方法，在之前写的获取/释放方法中调用AQS的acquire/release或者Shared方法

AQS在CountDownLatch中的应用

构造函数 发现，就是将count赋值给AQS中的成员变量state

public CountDownLatch(int count) { if (count < 0) throw new IllegalArgumentException("count < 0"); this.sync = new Sync(count); } Sync(int count) { setState(count); } protected final void setState(int newState) { state = newState; }

getCount 调用getCount方法，最终会获取到AQS里面的state

public long getCount() { return sync.getCount(); } int getCount() { return getState(); } protected final int getState() { return state; }

countDown

public void countDown() { sync.releaseShared(1); }

可以看到，countDown走的是释放共享锁的逻辑，从给state赋值也可以猜到用的是共享锁-有多个线程且state可赋大于0的值。继续看releaseShared逻辑：

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

可以看到就是读锁释放的逻辑，其中doReleaseShared方法实现逻辑相同就不看了，不同的是tryReleaseShared方法，下面跟进：

protected boolean tryReleaseShared(int releases) { // Decrement count; signal when transition to zero for (;;) { int c = getState(); if (c == 0) return false; int nextc = c-1; if (compareAndSetState(c, nextc)) return nextc == 0; } }

此方法在CountDownLatch中的内部类Sync中得到实现，逻辑为将state-1，并且如果是0的话返回true。返回true后在releaseShared方法中会进入if里面，走唤醒后续节点的逻辑doReleaseShared方法，在该方法中唤醒的main线程。main线程什么时候被挂起的？且看下面。

await

public void await() throws InterruptedException { sync.acquireSharedInterruptibly(1); }

await调用了可响应中断的获取共享锁方法，继续查看：

public final void acquireSharedInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); if (tryAcquireShared(arg) < 0) doAcquireSharedInterruptibly(arg); }

此方法是AQS中的公用模板方法，不同点在于各实现类的实现逻辑，在CountDownLatch中对tryAcquireShared方法进行了实现，实现逻辑如下：

protected int tryAcquireShared(int acquires) { return (getState() == 0) ? 1 : -1; }

即如果state==0则能获取到锁，否则获取不到。获取不到进入下面的doAcquireSharedInterruptibly方法，最终会将head的waitStatus设置为-1，自己挂起等待唤醒。

 

AQS在CountDownLatch中的总结 CountDownLatch是基于共享锁实现的并发控制功能，现在对总的实现逻辑做个梳理：

• 首先在构造器初始化CountDownLatch的时候，就会给AQS中的state赋值
• 调用await方法时便会尝试获取共享锁，不过一开始是获取不到锁的，于是线程阻塞。await方法是加锁的逻辑，但加锁条件是state==0时才会加锁成功，否则挂起；
• 而锁计数器的初始值为state，而后每一个线程调用一次countDown方法则共享锁释放一次，直到释放完；
• 最后，当通过countDown的调用将state减为0后，会唤醒处于阻塞状态的主线程，让其获取到锁并执行。

AQS在Semaphore中的应用

• 在Semaphore中，state表示许可证的剩余数量

Semaphore构造器

public Semaphore(int permits) { sync = new NonfairSync(permits); } public Semaphore(int permits, boolean fair) { sync = fair ? new FairSync(permits) : new NonfairSync(permits); }

可以看到，Semaphore有两个构造器，一个是只传数值默认非公平锁，另一个可指定用公平锁还是非公平锁。permits最终还是赋值给了AQS中的state变量。

acquire(int permits)

public void acquire(int permits) throws InterruptedException { if (permits < 0) throw new IllegalArgumentException(); sync.acquireSharedInterruptibly(permits); }

此方法同样调用了AQS中的模板方法：

public final void acquireSharedInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); if (tryAcquireShared(arg) < 0) doAcquireSharedInterruptibly(arg); }

• 1、查看tryAcquireShared的实现方法 先看非公平锁的获取：

protected int tryAcquireShared(int acquires) { return nonfairTryAcquireShared(acquires); } final int nonfairTryAcquireShared(int acquires) { for (;;) { int available = getState(); int remaining = available - acquires;// 如果remaining是负的，说明当前剩余的信号量不够了，需要阻塞 if (remaining < 0 || compareAndSetState(available, remaining))// 如果remaining<0则直接return，不会走CAS；如果大于0，说明信号量还够，可走CAS将信号量减掉，成功则返回大于0的remaining return remaining; } }

再看公平锁的获取

protected int tryAcquireShared(int acquires) { for (;;) { if (hasQueuedPredecessors())// 判断是不是在队首，不是的话直接返回-1 return -1; int available = getState();// 后面逻辑通非公平锁的获取逻辑 int remaining = available - acquires; if (remaining < 0 || compareAndSetState(available, remaining)) return remaining; } }

可以看到，不管非公平锁和公平锁，加锁时都是先判断当前state够不够减的，如果减出负数返回获取锁失败，是正数才走CAS将原信号量扣掉，返回获取锁成功。加锁时一个减state的过程。

• 2、doAcquireSharedInterruptibly 此方法还是AQS中的实现，逻辑重复，就不再说明了。

release(int permits)

public void release(int permits) { if (permits < 0) throw new IllegalArgumentException(); sync.releaseShared(permits); }

同样调用了AQS中的模板方法releaseShared：

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

其中tryReleaseShared的实现在Semaphore类的Sync中，如下所示：

protected final boolean tryReleaseShared(int releases) { for (;;) { int current = getState(); int next = current + releases;// 用当前state加上要释放的releases if (next < current) // overflow throw new Error("Maximum permit count exceeded"); if (compareAndSetState(current, next))// 用CAS将state加上 return true; } }

AQS在Semaphore中的总结 Semaphore信号量类基于AQS的共享锁实现，有公平锁和非公平锁两个版本。它的加锁与释放锁的不同之处在于和普通的加锁释放锁反着，ReentrantLock和ReentrantReadWriteLock中都是加锁时state+1，释放锁时state-1，而Semaphore中是加锁时state减，释放锁时state加。

 

另外，如果它还可以acquire(2) 、release(1)，即获取的和释放的信号量可以不一致，只是需要注意别释放的信号量太少导致后续任务获取不到足够的量而永久阻塞。

AQS在ReentrantLock中的应用

从下往上看，ReentrantLock类内部有两个静态内部类FairSync和NonfairSync，分别代表了公平锁和非公平锁（注意ReentrantLock实现的锁是可重入排它锁）。这两个静态内部类又共同继承了ReentrantLock的一个内部静态抽象类Sync，此抽象类继承AQS。

 

ReentrantLock的默认构造方法创建的是非公平锁，也可以通过传入true来指定生成公平锁。下面我们以公平锁的加锁过程为例，进行解读源码。在解读源码之前需要先明确一下AQS中的state属性，它是int类型，state=0表示当前lock没有被占用，state=1表示被占用，如果是重入状态，则重入了几次state就是几。

 

分析释放锁的方法tryRelease

• 由于是可重入的，所以state代表重入的次数，每次释放锁，先判断是不是当前持有锁的线程释放的，如果不是就抛异常；如果是的话，重入次数就减1，如果减到了0，就说明完全释放了，于是free就是true，并且把state设置为0

加锁的方法

• 会先判断当前state是不是为0，也会去判断当前线程是不是目前持有锁的线程，如果都不是代码目前获取不到这把锁，那么就把当前线程放入队列中去等待，并在以后合适的时机唤醒

ReentrantLock源码分析：

AQS系列（一）- ReentrantLock的加锁

AQS系列（二）- ReentrantLock的释放锁

AQS系列（三）- ReentrantReadWriteLock读写锁的加锁

AQS系列（四）- ReentrantReadWriteLock读写锁的释放锁

 

参考： https://www.cnblogs.com/fanBlog/p/9336126.html

https://blog.csdn.net/mulinsen77/article/details/84583716

https://www.cnblogs.com/iou123lg/p/9464385.html

https://www.cnblogs.com/zzq6032010/p/12076689.html

https://www.cnblogs.com/zzq6032010/p/12076687.html