C/C++教程
AQS学习（二） AQS互斥模式与ReenterLock可重入锁原理解析
1. MyAQS介绍
在这个系列博客中，我们会参考着jdk的AbstractQueuedLongSynchronizer，从零开始自己动手实现一个AQS（MyAQS）。通过模仿，自己造轮子来学习主要有两个好处，一是可以从简单到复杂，从核心逻辑再到旁路逻辑的实现，学习曲线较为平滑；二是可以站在设计者的角度去思考实现具体功能时可以采用的各种方案的优劣，更深刻的理解其设计的精妙、关键之处。
AQS支持互斥和共享这两种工作模式，其中互斥模式比共享模式要简单许多。本着由浅入深的原则，本篇博客实现的MyAQS暂时只支持互斥模式。
MyAQS会按照顺序，逐步的实现互斥模式、共享模式、允许取消加锁（中断、超时退出）和支持条件变量这四个模块，具体内容会在后续的博客中分享出来。
1.1 乐观锁与CAS原理介绍
基于CAS策略的乐观锁机制是实现无锁并发的关键所在，因此在展开AQS的实现原理前需要先简单介绍一下乐观锁和CAS机制的原理。
悲观锁与乐观锁是用来控制线程并发安全的两种机制。为了防止临界区数据被不同线程并发的读写出现问题，悲观锁只允许一个线程进入临界区以进行临界区数据的访问，而其余没有争用到锁的线程则会停留在临界区外（自旋或者进入阻塞态）。而乐观锁则是基于比较并设置这一思想来实现的，其允许不同线程并发的访问、修改某一临界区数据，但保证同一瞬间只有一个线程能够修改成功。从这个角度看乐观锁其实并不是传统概念上的锁，而更像是一种策略。
具体来说，乐观锁中每个线程在需要修改某一临界区数据前需要先读取当前数据的快照值（expect），然后执行一次cas操作（compareAndSet），如果CAS操作返回成功则说明修改成功，如果返回失败则说明对应数据在当前线程读取快照后、执行cas操作前的这段时间内有别的线程已经进行过修改，则需要重新读取出当前最新的快照值进行处理后再次尝试cas操作。
并发场景下cas操作可能会失败很多次，所以一般是放在一个循环中执行的，无限循环直到cas操作成功才结束。
CAS操作示例伪代码（CAS自增）：
boolean compareAndSet(expect,update,targetDataMemory){
if(expect == targetDataMemory.data){ // compare 比较
targetDataMemory.data = update; // set 设置
return true;
}else{
return false;
}
}
Integer atomicIntegerAdd(){
// targetDataMemory.data标识对应的整数
do{
expect = targetDataMemory.data;
newUpdate = expect+1;
}while(compareAndSet(expect, newUpdate, targetDataMemory));
return newUpdate;
}
从伪代码中可以看到，compareAndSet中有一次比较操作（expect == targetDataMemory.data）和一次赋值操作（targetDataMemory.data = update），如果在比较和赋值操作中出现了并发（线程A执行比较为true，线程B执行比较也为true，线程B执行赋值，线程A再执行赋值），则线程A将会覆盖掉线程B的改动，那这个cas操作就是有问题的。因此cas操作作为一个基础的底层操作，其对数据的比较和修改必须是原子性的，即compare和set是一起执行的，在此期间不允许插入其它操作。
保证原子性操作的手段有很多，在应用程序层可以实现，在操作系统层、硬件层也可以实现。但由于上层功能的实现都是基于底层功能的，如果没有硬件层提供的原子操作，是不可能在基于硬件上的软件层实现原子操作的，而compareAndSet操作由于其指令集合足够简单，因此很自然的被cpu硬件实现了，使得可以通过一条汇编指令来控制cpu去执行cas指令。基于硬件提供的各种基础的原子性操作，上层的操作系统、应用程序才能够实现更高级的阻塞/唤醒、信号量、互斥锁等等功能。
在java中提供了许多可以利用硬件CAS机制的工具类，例如juc包下的atomic系列工具类集合（AtomicInteger、AtomicReference等)；unsafe类中提供的cas方法（compareAndSwapInt、compareAndSwapObject等）。
1.2 MyAQS基础结构
MyAQS是jdkAQS的简化版，因此整体的结构与jdk中的AQS差别不大，都有一个双向链表结构的同步队列作为底层支撑。由于本篇博客中的AQS只实现了互斥模式，并且暂时还未引入jdk实现中那么多处理取消加锁的复杂逻辑和节点状态(status)，代码相对简单很多，也能够更加准确的把握住互斥模式中最核心的逻辑。
下面开始介绍MyAQS第一版的基础结构。
1. 同步队列头、尾节点以及相关的CAS操作封装
AQS的同步队列是基于有显式前驱结点引用的CLH锁队列的一个变种实现（理解CLH锁工作原理对后续理解AQS的实现会有很大帮助，可以参考我之前的博客AQS学习（一）自旋锁原理介绍）
和CLH锁一样，AQS的底层队列同样有一个虚拟的Dummy头节点（head），以及一个标识当前队尾节点的tail节点。由于需要实现队列的无锁并发，因此head节点与tail节点都是使用volatile关键字修饰的，并且在会出现并发的临界区代码中使用CAS操作+乐观重试的机制来保证队列并发访问时的线程安全。
相关方法为：compareAndSetHead和compareAndSetTail。
2. 需要上层同步器实现的个性化尝试加锁、尝试解锁抽象方法
AQS作为一个底层的框架，是无法在内部同时去为各种类型的同步器实现线程何为加锁成功/失败的逻辑的，因此AQS将这些个性化的逻辑提取为了几个抽象方法交给特定的同步器按照约定去实现，而AQS的内部会在需要的时候调用这些方法，这也是AQS被定义为抽象类的原因。
其中在互斥模式下需要子类去实现两个方法：tryAcquire和tryRelease。
- tryAcquire用于尝试着去争用互斥锁，约定加锁成功返回true、加锁失败返回false。
- tryRelease用于尝试着去释放释放锁，约定解锁成功返回true，解锁失败返回false。
这两个方法具体的使用会在下面的互斥模式工作原理中展开介绍。
3 提供给上层同步器使用的state字段
AQS提供了一个使用volatile关键字修饰的int类型属性state，特定的同步器可以利用state属性来灵活的实现自己的需求。
相关方法为：getState、setState和compareAndSetState（CAS设置state的值）。
MyAQS基础结构代码：
public abstract class MyAqsV1 {
private volatile int state;
private transient volatile Node head;
private transient volatile Node tail;
private transient Thread exclusiveOwnerThread;
private static final Unsafe unsafe;
private static final long stateOffset;
private static final long headOffset;
private static final long tailOffset;
static {
try {
// 由于提供给cas内存中字段偏移量的unsafe类只能在被jdk信任的类中直接使用，这里使用反射来绕过这一限制
Field getUnsafe = Unsafe.class.getDeclaredField("theUnsafe");
getUnsafe.setAccessible(true);
unsafe = (Unsafe) getUnsafe.get(null);
stateOffset = unsafe.objectFieldOffset(MyAqsV1.class.getDeclaredField("state"));
headOffset = unsafe.objectFieldOffset(MyAqsV1.class.getDeclaredField("head"));
tailOffset = unsafe.objectFieldOffset(MyAqsV1.class.getDeclaredField("tail"));
} catch (Exception var1) {
throw new Error(var1);
}
}
/**
* 设置独占当前aqs的线程
* */
protected final void setExclusiveOwnerThread(Thread thread) {
exclusiveOwnerThread = thread;
}
/**
* 获取独占当前aqs的线程
* */
protected final Thread getExclusiveOwnerThread() {
return exclusiveOwnerThread;
}
protected final int getState() {
return state;
}
protected final void setState(int newState) {
state = newState;
}
protected final boolean compareAndSetState(int expect, int update) {
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}
private boolean compareAndSetHead(Node update) {
return unsafe.compareAndSwapObject(this, headOffset, null, update);
}
private boolean compareAndSetTail(Node expect, Node update) {
return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
}
private void setHead(Node node) {
head = node;
node.thread = null;
node.prev = null;
}
/**
* 尝试着去申请互斥锁（抽象方法，由具体的实现类控制）
* */
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}
/**
* 尝试着去释放互斥锁（抽象方法，由具体的实现类控制）
* */
protected boolean tryRelease(int arg) {
throw new UnsupportedOperationException();
}
}
2. MyAQS同步队列
2.1 MyAQS队列节点
MyAQS队列节点中的成员属性可能会被多个线程并发的访问，所以需要用volatile关键字修饰以保证其在不同线程间内存的可见性。
最初的版本中节点属性比较简单，后续随着所要支持功能的增多，MyAQS中Node节点类也会随之拓展而变得复杂。
AQS节点类实现：
/**
* 内部同步队列的节点
* */
static final class Node {
volatile Node prev;
volatile Node next;
volatile Thread thread;
public Node() {
}
Node(Thread thread) {
this.thread = thread;
}
final Node predecessor() throws NullPointerException {
Node p = prev;
if (p == null) {
throw new NullPointerException();
}
else {
return p;
}
}
}
MyAQS同步队列示意图：
2.2 AQS队列入队
AQS作为互斥锁等同步器的底层支撑框架，当线程争用锁失败时需要令当前线程节点插入同步队列的队尾，入队之后线程会陷入阻塞而让出CPU，等待其前驱节点在释放锁后将其唤醒。
AQS队列入队相关实现：
/**
* 尝试加互斥锁，如果加锁失败则当前线程进入阻塞状态
* */
public final boolean acquire(int arg) {
// 尝试着去申请互斥锁
boolean acquireResult = tryAcquire(arg);
if(!acquireResult){
// 申请互斥锁失败，新创建一个绑定当前线程的节点，并令其插入队尾
Node newWaiterNode = addWaiter();
// 尝试着加入同步队列
acquireQueued(newWaiterNode,arg);
}
// 不支持中断功能，返回false
return false;
}
/**
* 创建当前线程对应的同步队列节点
* 令该队列节点插入队尾
* */
private Node addWaiter() {
Node node = new Node(Thread.currentThread());
Node pred = tail;
if (pred != null) {
node.prev = pred;
// 当队列不为空时，执行一次快速的入队操作（因为少了一次enq方法调用，会快一点？）
if (compareAndSetTail(pred, node)) {
// 快速入队成功，直接返回
pred.next = node;
return node;
}
}
// 上面的快速入队操作失败了，使用enq循环cas直到入队（队列为空，利用enq方法初始化同步队列）
enq(node);
return node;
}
/**
* 入队操作
* 使用CAS操作+无限重试的方式来解决并发冲突的问题
* @return 返回新的队尾节点
* */
private Node enq(final Node node) {
for (;;) {
Node currentTailNode = tail;
if (currentTailNode == null) {
// AQS的同步队列是惰性加载的，如果tail为null说明队列为空（head=null && tail=null）
if (compareAndSetHead(new Node())) {
// 使用cas的方式先创建一个新节点，令tail和head同时指向这一新节点
// 并发时多个线程同时执行，只会有一个线程成功执行compareAndSetHead这一cas操作
tail = head;
}
} else {
// 令当前入队节点node插入队尾
node.prev = currentTailNode;
// 使用cas的方式令aqs的tail指向node，使得新节点node成为新的队尾元素
if (compareAndSetTail(currentTailNode, node)) {
// 并发时多个线程同时执行，获取到的tail引用值是一样的，只有最先执行compareAndSetTail的线程会成功
// compareAndSetTail执行成功后令tail引用指向了一个新的节点，因此同一时刻获取到相同tail引用的线程cas插入队尾的操作会失败（expect不对了）
currentTailNode.next = node;
return currentTailNode;
}
// compareAndSetTail执行失败的线程会进入新的循环，反复尝试compareAndSetTail的cas操作直到最终成功
}
}
}
/**
* 令已经入队后的节点陷入阻塞态
* */
private void acquireQueued(final Node node, int arg) {
for (; ; ) {
final Node p = node.predecessor();
// 如果需要当前节点是aqs头节点的next节点，则尝试tryAcquire获取锁
if (p == head && tryAcquire(arg)) {
// tryAcquire获取锁成功成功，说明头节点对应的线程已经释放了锁
// 令当前入队的节点成为新的head节点
setHead(node);
p.next = null; // help GC
return;
}else{
// 阻塞当前线程
LockSupport.park(this);
}
}
}
上述入队相关逻辑中有几个需要重点关注的逻辑：
2.2.1 acquire入队总控方法
对外提供的public方法acquire是AQS互斥模式下加锁的总控方法。当子类实现的tryAcquire尝试加锁失败而返回false时，会先调用addWaiter方法获得已入队的线程节点，再调用acquireQueued方法令对应线程陷入阻塞态。
2.2.2 创建同步队列节点并入队
addWaiter方法用于将当前线程节点置入队尾。而其中head节点是懒加载的，即只有当第一个线程节点需要入队时才会被初始化。这样做的主要好处是当AQS中只有一个线程占用锁，但不存在其它争用锁失败的线程时，通过推迟head节点的创建时机而提高AQS的空间利用率。
2.2.3 CAS的无锁并发队列
由于是无锁并发的队列，addWaiter和enq方法中如果存在多个线程同时并发的入队并通过CAS设置tail节点引用时，至多只会有一个线程CAS返回成功，而其它线程则会CAS失败。
失败的线程由于处于for的无限循环中，会重新读取新的tail节点引用值并再次尝试着CAS加入队尾（因为之前成功的节点已经成为了新的队尾节点，当前线程需要挂在最新的队尾节点之后）。在多个线程并发操作时，部分的线程可能会失败很多次，但只要并发的线程数不再持续增加，最终所有的入队CAS操作都将成功。
2.2.4 当前入队节点其前驱节点不是头节点时，入队后会陷入阻塞态
一般的，争用锁失败而加入同步队列成功的线程在acquireQueued中会通过LockSupport.park方法而陷入阻塞态，等待前驱节点在释放锁时将其重新唤醒。
2.2.5 当前节点前驱节点恰好为头节点时，陷入阻塞态前先尝试一次tryAcquire
特别的，如果acquireQueued中当前节点node其前驱恰好是头节点，则会再尝试一次tryAcquire方法，若此时子类实现的tryAcquire返回true则认为加锁成功。加锁成功后，当前线程不必进入阻塞态而是直接返回，返回前会通过setHead方法令自己成为新的线程节点。setHead和p.next = null这两行代码会将之前老的head节点相关引用全部设置为null，之后老的head节点便会被自动GC给回收掉。
虽然在总控函数acquire中已经先执行过了一次tryAcquire，并且当时返回false时才会执行到acquireQueued方法中。但由于多线程并发时存在非常多的临界情况，头节点所对应的线程可能在当前线程执行到acquireQueued时就已经释放了锁，此时再次执行tryAcquire便会返回true了。
2.3 AQS队列出队
AQS和CLH锁在线程释放锁时的行为略有不同。
CLH锁在释放锁时只是简单的将当前节点的isLocked锁标记修改为false，如果此时后继节点存在的话，就会在自旋循环中及时的感知到这一变化后而加锁成功。
但AQS中加锁失败并入队的后继节点一般是位于阻塞态的，无法主动发现这一变化，因此需要由释放锁的线程负责去将其唤醒。这也是为什么AQS的队列要在CLH锁的基础上引入显式next后继节点引用的原因（CLH锁队列的变种）。因为这样可以在大多数情况下将获取后继节点引用的时间复杂度由O(n)（从tail队尾通过prev引用一个一个找过来）优化为O(1)（head.next即可）。
AQS队列出队相关实现：
/**
* 尝试着释放互斥锁
* @return true：释放成功；false：释放失败
* */
@Override
public final boolean release(int arg) {
// 尝试着释放锁
if (tryRelease(arg)) {
// 成功释放
Node h = this.head;
if (h != null) {
unparkSuccessor(h);
}
return true;
}
return false;
}
/**
* 唤醒后继节点
* */
private void unparkSuccessor(Node node) {
Node next = node.next;
if(next != null) {
LockSupport.unpark(next.thread);
}
}
对外提供的public方法release是AQS互斥模式下释放锁的总控方法。当子类实现的tryRelease尝试解锁成功而返回true时，若当前同步队列head头节点存在，且头节点的next后继也存在，则将头节点的直接后继节点对应的线程唤醒。
2.4 AQS多线程入队/出队时的临界状态分析
将加锁失败入队阻塞的acquireQueued方法和解锁成功时唤醒后继的unparkSuccessor方法结合起来就会发现，大多数时候互斥模式下head节点对应线程在释放锁时，会通过LockSupport.unpark方法将此前被LockSupport.park阻塞的后继节点唤醒。被唤醒的后继节点由于处于循环中，会再度尝试tryAcquire方法，如果返回true则加锁成功成为新的头节点；如果返回false则将被LockSupport.park再次阻塞，等待被唤醒。
但在上述场景之外还存在几个微妙的临界状态需要仔细分析。
1. 加锁失败的线程节点还未完全入队时，恰好此时head节点对应线程释放了互斥锁
已经获取到锁的线程A在释放互斥锁时，争用锁失败的线程B正在入队（假设当前只有A、B两个线程存在），但其前驱head头节点还未与之建立next关联（线程B的节点已经成为新的队尾节点但在addWaiter中的pred.next = node执行前）。此时线程A执行release方法，发现头节点的next引用不存在，因此不会去执行LockSupport.unpark方法。但只要子类在tryRelease中正确的进行了处理，那么将不会出现lost wakeup问题。
具体来说，当线程B执行acquireQueued时，tryAcquire要么返回true，线程B成功加锁而无需陷入阻塞态；或者非公平模式下有别的线程C已经抢先获得了锁，那么线程B节点执行accuireQueued时已经是head节点的直接后继了，tryAcquire由于C已经抢险加锁成功而返回false后线程B陷入阻塞，当抢先获得锁的线程C释放锁时执行release方法时便能正确的将线程A唤醒。
2. 入队线程执行acquiredQueued与持有锁的线程解锁时unparkSuccessor并发执行时
现在考虑这样一种场景，争用锁线程A调用acquire方法时加锁失败，其线程节点作为head头节点的直接后继已经入队完成，并且已经执行到acquireQueued方法内部，正准备执行LockSupport.park将自己阻塞（还未执行park方法，且此时head.next = NodeA）；另一方面之前持有锁的线程B正在通过release释放锁，在tryRelease返回true后，准备通过unparkSuccessor中的LockSupport.unpark将后继唤醒。
那么如果在线程A执行LockSupport.park(this)阻塞自己之前，线程B先一步执行了LockSupport.unpark(next.thread)，线程A会不会出现lost wakeup问题呢？
答案是不会的，LockSupport.park/unpark一般成对的使用用于线程同步，其内部有一个基于线程级别的、允许预先设置的许可标记，且默认情况下许可是不存在的。
LockSupport.park时如果发现许可不存在会令当前线程陷入阻塞态；若许可已存在则不必阻塞直接返回，并消费掉许可位（再次执行的话就会被阻塞了）。而LockSupport.unpark(thread)时如果发现对应线程thread此时没有因为执行了LockSupport.park陷入阻塞，则会为参数线程thread预先设置一个许可；若对应线程已经先执行了LockSupport.park，则会将线程从阻塞态唤醒。
若线程A先LockSupport.park则会先陷入阻塞，线程B随后会通过LockSupport.unpark(thread)将线程A唤醒；若线程B先执行LockSupport.unpark(thread)，则会为线程A预先设置许可，当线程A后执行LockSupport.park时发现许可已存在，则消费掉许可并直接返回而不会陷入阻塞。
综上所属，无论并发时线程A的线程是先于线程B调用LockSupport.unpark(thread)前调用LockSupport.park还是之后调用，都不会有问题。
2.5 AQS提供的hasQueuedPredecessors方法
对于需要先来先服务的公平锁，如果之前同步队列中已经存在其它需要争用锁的线程节点，则当前加锁线程作为后来者将不能先于之前的线程获得锁。通过AQS提供的hasQueuedPredecessors方法就能很简单的辅助同步器实现先来先服务的公平锁机制。
hasQueuedPredecessors实现：
public final boolean hasQueuedPredecessors() {
Node t = tail;
Node h = head;
if(h == t){
// tail=head两种情况，都为null说明队列为空；或者都指向同一个节点，队列长度为1
// 说明此时队列中并没有其它等待锁的线程，返回false
return false;
}
Node secondNode = h.next;
if(secondNode != null){
if(secondNode.thread == Thread.currentThread()){
// 头节点存在后继节点，且后继节点就是当前线程自己，因此不需要排队
return false;
}else{
// 头节点存在后继节点，但后继节点不是当前线程，因此需要排队
return true;
}
}else{
// tail != head，但是头节点却没有next节点，这是一种特殊的场景
// 在enq入队操作的初始化队列操作时可能会出现，先通过compareAndSetHead设置了头节点，但是还没执行tail = head操作前的瞬间会出现
// 此时，说明已经有一个别的线程正在执行入队操作，而当前线程此时还未进行入队，相对进度更慢，所以还是需要去排队的
return true;
}
}
熟悉jdk AQS源码的读者可能会注意到上述代码形式于jdk中的代码结构不太一样，但实际上这里只是将jdk源码中的代码在保持逻辑不变的情况下进行了适当改写，使得各个判断的逻辑分支更为清晰而更易理解。
3. ReenterLock可重入互斥锁工作原理
在理解了AQS的同步队列工作原理后，AQS互斥模式的工作原理就比较容易理解了。对于一个基于AQS互斥模式的同步锁，AQS要做的就是在线程争用锁失败时为对应线程创建一个线程节点并使其插入队尾，同时令其暂时进入阻塞态以等待之前已获得锁的线程在释放锁之后将其唤醒。
虽然已经进行了关于AQS的同步队列与互斥模式工作原理的介绍，但AQS抽象同步队列无愧于抽象之名，光是站在设计者的角度对内部各个模块进行拆解学习是不够的，还需要站在使用者的角度将上面关于AQS的各个模块串联起来，从AQS的外部以一个更全面的视角来学习和加深对其的理解。而同样位于juc包下的可重入互斥锁ReenterLock就是一个基于AQS互斥模式，非常常见的同步器。通过对ReenterLock实现的学习，将其与AQS的工作原理结合起来，看看ReenterLock是如何基于AQS框架实现互斥、可重入、公平/非公平等特性的。
为了更好地测试我们实现的AQS，这里将ReenterLock的源码进行一定的简化(MyReenterLock)，暂时去除了条件变量Condition等内容，仅保留了最基本的互斥加锁功能，同时将使用到jdk中AQS的地方替换成我们自己实现的MyAQS版本。
MyReenterLock类实现：
/**
* 将jdk的ReentrantLock中的aqs改成MyAqsV1
*/
public class MyReentrantLockV1 {
private final MyReentrantLockV1.Sync sync;
public MyReentrantLockV1(boolean fair) {
sync = fair ? new FairSync() : new NonfairSync();
}
abstract static class Sync extends MyAqsV1 {
abstract void lock();
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
protected final boolean tryRelease(int releases) {
int c = getState() - releases;
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
if (c == 0) {
free = true;
setExclusiveOwnerThread(null);
}
setState(c);
return free;
}
}
static final class NonfairSync extends Sync {
/**
* Performs lock. Try immediate barge, backing up to normal
* acquire on failure.
*/
final void lock() {
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
}
static final class FairSync extends Sync {
final void lock() {
acquire(1);
}
/**
* Fair version of tryAcquire. Don't grant access unless
* recursive call or no waiters or is first.
*/
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
if (!hasQueuedPredecessors() &&
compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
}
public void lock() {
sync.lock();
}
public void unlock() {
sync.release(1);
}
}
juc包中并没有对同步器的API做一个统一的定义，因此不同的同步器可以有不同的api，所以AQS在ReenterLock中是以组合，而不是被直接继承的形式使用的。ReenterLock内部抽象出了一个叫做Sync的类用于继承并实现AQS中的抽象方法，将AQS的acquire、release等较为抽象的方法名给封装起来，以更加符合语义的方法名lock、unlock对外暴露接口。
ReenterLock中利用AQS提供的state属性判断当前是否有线程已经获得互斥锁，以及存储重入加锁的次数。具体来说，state=0代表当前没有线程持有该互斥锁，state>0时代表当前已经有线程持有了锁；ReenterLock是可重入的，持有锁的线程在每次加锁成功时state自增1，解锁时state自减1，当state减至0时便代表释放了锁。因此ReenterLock在使用时，需要如果线程多次调用lock加锁成功时，也需要调用同样次数的unlock方法才能成功释放锁。
ReenterLock支持以公平互斥锁和非公平互斥锁两种模式工作，可以通过构造方法传入boolean类型参数指定。
3.1 ReenterLock公平锁工作原理
公平模式下的ReenterLock是由FairSync来实现sync抽象类的。
公平模式下FairSync加锁
ReenterLock公平模式下lock加锁时执行sync.lock，实际上是执行子类FairSync的lock方法。FairSync的lock实现中会直接调用AQS的acquire方法，并且传入参数1（acquire(1)）。而AQS的acquire方法中将arg参数1透传给由子类FairSync实现的tryAcquire方法。
FairSync的tryAcquire实现中，首先判断state的值.
如果state为0，则先通过AQS的hasQueuedPredecessors判断当前同步队列是否已经存在其它的线程节点。作为公平锁hasQueuedPredecessors如果返回true，则tryAcquire直接返回false尝试加锁失败；如果hasQueuedPredecessors返回false，则进一步通过CAS的方式尝试着将state由0变为1，如果CAS成功则代表加锁成功，CAS失败则代表有其它线程也在这个瞬间并发的进行了加锁操作且CAS设置state由0到1成功，则当前线程也是tryAcquire加锁失败返回false。
如果state不为0，说明当前ReenterLock已经有线程持有锁了，则进一步的通过current == getExclusiveOwnerThread()判断当前加锁的线程是否就是持有锁的线程，如果加锁线程不是持有锁的线程，基于互斥性则加锁失败tryAcquire返回false；而如果当前加锁的线程正是持有锁的线程，则将当前state自增1，返回true代表加锁成功。
加锁失败时tryAcquire返回false，则会接着执行AQS的acquire方法中后续的线程节点入队和线程阻塞逻辑（addWaiter、acquireQueued）。
公平模式下解锁
ReenterLock公平模式下解锁时执行sync.unlock方法，实际上是执行父类中AQS的release方法。AQS的release方法传入arg参数1，release方法中调用并透传arg参数1给由子类实现的tryRelease方法。ReenterLock中sync类实现了tryRelease方法中，先将当前state减去1(arg参数)，若发现减去1之后的state为0，则代表当前线程已经成功释放了锁，tryRelease返回true；若state减去1之后state依然大于0，说明当前可重入锁还不能释放锁，tryRelease返回false。
如果解锁时tryRelease了返回true，则AQS的release方法中会接着唤醒当前持有锁节点的（head节点）直接后继节点（unparkSuccessor)。
3.2 ReenterLock非公平锁工作原理
非公平模式下的ReenterLock是由NonFairSync来实现sync抽象类的。
非公平模式下NonFairSync加锁
与公平模式下FairSync的实现不同的是，非公平锁下线程抢占锁时不需要考虑公平性，即加锁时不需要去考虑当前是否已经有别的线程先一步等待加锁。
因此NonFairSync的lock实现中首先尝试着通过CAS的方式去尝试着获得锁（compareAndSetState(0,1)），如果CAS操作成功则直接令当前线程获得锁（setExclusiveOwnerThread(Thread.currentThread())）。而如果CAS操作失败则和FairSync一样会去调用AQS的acquire方法，并且传入参数1（acquire(1)）。而AQS的acquire方法中将arg参数1透传给由子类NonFairSync实现的tryAcquire方法。
NonFairSync的tryAcquire实现和公平锁中的FairSync基本一致，唯一的区别在于当发现state为0时NonFairSync不会通过hasQueuedPredecessors校验队列中是否已存在其它等待获得锁的线程，而是允许当前线程直接通过CAS操作尝试争抢锁。因此非公平模式下，当之前获取到锁的线程通过unlock释放锁将state设置为0时，即使当前队列中已经存在其它被阻塞等待被唤醒的线程时，新来争用锁的线程也可能抢在队列中线程被唤醒并拿到锁之前抢先一步获得锁（state+1）。此时被唤醒的线程接着执行acquireQueued方法时再一次尝试tryAcquire，因为state不为0而返回失败后便又会继续陷入阻塞态。
非公平模式下解锁
ReenterLock非公平模式下的解锁与公平模式下的解锁逻辑完全一致，不再赘述。
3.3 ReenterLock公平锁与非公平锁的差别
- 从功能上来说，ReenterLock的公平锁和非公平锁的主要区别在于非公平锁允许后申请争用锁的线程抢先获得锁，而公平锁不能。
- 从代码实现上来说，非公平模式下允许争抢锁的时机主要有两处：一是在lock时允许尝试一次CAS操作进行争抢（NonFairSync.lock方法）；二是在tryAcquire时如果发现state为0，允许当前线程再一次的通过CAS尝试争抢（NonFairSync.tryAcquire方法）。但是非互斥模式下加锁时若这两次争用都未能成功获取到锁，则当前线程依然会和公平锁模式下一样乖乖的加入同步队列，等待之前已在队列中的线程依次释放锁后来将其唤醒。
-
从性能上来说，由于头节点线程释放锁时，唤醒队列中被阻塞的线程时由于涉及到操作系统中线程的上下文切换而存在一定的延迟。若此时能允许新的线程在加锁时抢先获得锁去执行被互斥锁保护的临界区逻辑，则总体的加锁/解锁吞吐量会较之不允许抢占的策略更高。因此非公平锁的性能会略高于公平锁，但在并发较高的场景下新加锁的线程可能会频繁的抢先，导致已存在于队列中的线程长时间无法获得锁而出现饥饿问题。
总结
MyAQSV1完整实现：
import sun.misc.Unsafe;
import java.lang.reflect.Field;
import java.util.concurrent.locks.LockSupport;
/**
* 自己实现的aqs，v1版本
* 只支持互斥锁模式（无法处理被阻塞线程发生被中断）
*/
public abstract class MyAqsV1 implements MyAqs {
private volatile int state;
private transient volatile Node head;
private transient volatile Node tail;
private transient Thread exclusiveOwnerThread;
private static final Unsafe unsafe;
private static final long stateOffset;
private static final long headOffset;
private static final long tailOffset;
static {
try {
// 由于提供给cas内存中字段偏移量的unsafe类只能在被jdk信任的类中直接使用，这里使用反射来绕过这一限制
Field getUnsafe = Unsafe.class.getDeclaredField("theUnsafe");
getUnsafe.setAccessible(true);
unsafe = (Unsafe) getUnsafe.get(null);
stateOffset = unsafe.objectFieldOffset(MyAqsV1.class.getDeclaredField("state"));
headOffset = unsafe.objectFieldOffset(MyAqsV1.class.getDeclaredField("head"));
tailOffset = unsafe.objectFieldOffset(MyAqsV1.class.getDeclaredField("tail"));
} catch (Exception var1) {
throw new Error(var1);
}
}
/**
* 设置独占当前aqs的线程
* */
protected final void setExclusiveOwnerThread(Thread thread) {
exclusiveOwnerThread = thread;
}
/**
* 获取独占当前aqs的线程
* */
protected final Thread getExclusiveOwnerThread() {
return exclusiveOwnerThread;
}
protected final int getState() {
return state;
}
protected final void setState(int newState) {
state = newState;
}
protected final boolean compareAndSetState(int expect, int update) {
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}
/**
* 尝试加互斥锁，如果加锁失败则当前线程进入阻塞状态
* */
@Override
public final boolean acquire(int arg) {
// 尝试着去申请互斥锁
boolean acquireResult = tryAcquire(arg);
if(!acquireResult){
// 申请互斥锁失败，新创建一个绑定当前线程的节点
Node newWaiterNode = addWaiter();
// 尝试着加入同步队列
acquireQueued(newWaiterNode,arg);
}
// 不支持中断功能，返回false
return false;
}
/**
* 尝试着释放互斥锁
* @return true：释放成功；false：释放失败
* */
@Override
public final boolean release(int arg) {
// 尝试着释放锁
if (tryRelease(arg)) {
// 成功释放
Node h = this.head;
if (h != null) {
unparkSuccessor(h);
}
return true;
}
return false;
}
@Override
public final boolean hasQueuedPredecessors() {
Node t = tail;
Node h = head;
if(h == t){
// tail=head两种情况，都为null说明队列为空；或者都指向同一个节点，队列长度为1
// 说明此时队列中并没有其它等待锁的线程，返回false
return false;
}
Node secondNode = h.next;
if(secondNode != null){
if(secondNode.thread == Thread.currentThread()){
// 头节点存在后继节点，且就是当前线程，因此不需要排队
return false;
}else{
// 头节点存在后继节点，但不是当前线程，因此需要排队
return true;
}
}else{
// tail != head，但是头节点却没有next节点，这是一种特殊的场景
// 在enq入队操作的初始化队列操作时可能会出现，先通过compareAndSetHead设置了头节点，但是还没执行tail = head操作前的瞬间会出现
// 此时，说明已经有一个别的线程正在执行入队操作，当前线程此时还未进行入队，进度更慢，所以还是需要去排队的
return true;
}
}
/**
* 尝试着加入队列
* */
private void acquireQueued(final Node node, int arg) {
for (; ; ) {
final Node p = node.predecessor();
// 如果需要入队的节点是aqs头节点的next节点，则最后尝试一次tryAcquire获取锁
// 这里的判断有两个作用
// 1 当前线程第一次执行acquireQueued还未被LockSupport.park阻塞前，若当前线程的前驱恰好是头节点则
// 最后再通过tryAcquire判断一次，若恰好这个临界点上头节点对应的线程已经释放了锁，则可以免去一次LockSupport.park
// 2 当前线程已经不是第一次执行acquireQueued，而是已经至少被LockSupport.park阻塞过一次
// 则在被前驱节点唤醒后在for的无限循环中通过tryAcquired再尝试一次加锁
// 若是公平锁模式下，则此时tryAcquire应该会返回true而加锁成功return退出
// 若是非公平锁模式下，若此时有别的线程抢先获得了锁，则tryAcquire返回false，当前被唤醒的线程再一次通过LockSupport.park陷入阻塞
if (p == head && tryAcquire(arg)) {
// tryAcquire获取锁成功成功，说明此前的瞬间头节点对应的线程已经释放了锁
// 令当前入队的节点成为aqs中新的head节点
setHead(node);
p.next = null; // help GC
return;
}else{
// 阻塞当前线程
LockSupport.park(this);
}
}
}
/**
* 唤醒后继节点
* */
private void unparkSuccessor(Node node) {
Node next = node.next;
if(next != null) {
LockSupport.unpark(next.thread);
}
}
private void setHead(Node node) {
head = node;
node.thread = null;
node.prev = null;
}
/**
* 创建当前线程对应的同步队列节点
* 令该队列节点插入队尾
* */
private Node addWaiter() {
Node node = new Node(Thread.currentThread());
Node pred = tail;
if (pred != null) {
node.prev = pred;
// 当队列不为空时，执行一次快速的入队操作（因为少了一次enq方法调用，会快一点？）
if (compareAndSetTail(pred, node)) {
// 快速入队成功，直接返回
pred.next = node;
return node;
}
}
// 上面的快速入队操作失败了，使用enq循环cas直到入队（队列为空，利用enq方法初始化同步队列）
enq(node);
return node;
}
/**
* 入队操作
* 使用CAS操作+无限重试的方式来解决并发冲突的问题
* @return 返回新的队尾节点
* */
private Node enq(final Node node) {
for (;;) {
Node currentTailNode = tail;
if (currentTailNode == null) {
// AQS的同步队列是惰性加载的，如果tail为null说明队列为空（head=null && tail=null）
if (compareAndSetHead(new Node())) {
// 使用cas的方式先创建一个新节点，令tail和head同时指向这一新节点
// 并发时多个线程同时执行，只会有一个线程成功执行compareAndSetHead这一cas操作
tail = head;
}
} else {
// 令当前入队节点node插入队尾
node.prev = currentTailNode;
// 使用cas的方式令aqs的tail指向node，使得新节点node成为新的队尾元素
if (compareAndSetTail(currentTailNode, node)) {
// 并发时多个线程同时执行，获取到的tail引用值是一样的，只有最先执行compareAndSetTail的线程会成功
// compareAndSetTail执行成功后令tail引用指向了一个新的节点，因此同一时刻获取到相同tail引用的线程cas插入队尾的操作会失败（expect不对了）
currentTailNode.next = node;
return currentTailNode;
}
// compareAndSetTail执行失败的线程会进入新的循环，反复尝试compareAndSetTail的cas操作直到最终成功
}
}
}
/**
* 尝试着去申请互斥锁（抽象方法，由具体的实现类控制）
* */
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}
/**
* 尝试着去释放互斥锁（抽象方法，由具体的实现类控制）
* */
protected boolean tryRelease(int arg) {
throw new UnsupportedOperationException();
}
/**
* 内部同步队列的节点
* */
static final class Node {
volatile Node prev;
volatile Node next;
volatile Thread thread;
public Node() {
}
Node(Thread thread) {
this.thread = thread;
}
final Node predecessor() throws NullPointerException {
Node p = prev;
if (p == null) {
throw new NullPointerException();
}
else {
return p;
}
}
}
private boolean compareAndSetHead(Node update) {
return unsafe.compareAndSwapObject(this, headOffset, null, update);
}
private boolean compareAndSetTail(Node expect, Node update) {
return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
}
}
View Code
本篇博客通过模仿jdk的AQS，实现了一个暂时只支持互斥模式的简易版AQS，即MyAQSV1。MyAQSV1在实现功能的基础上，尽可能的裁剪了jdk实现中与互斥功能不相关的逻辑，以一个最小子集的形式工作；同时也对jdk实现中一些不易理解的地方进行了改写使之更易理解。MyAQSV1从内部解析了AQS互斥模式的工作原理，而简易版的互斥锁MyReenterLock则让我们能够以外部使用者的角度来观察AQS的行为。
希望这篇博客能够帮助读者更好的理解AQS中无锁并发的同步队列和互斥模式下争用锁和释放锁的机制，以及ReenterLock中公平锁与非公平锁具体实现上和性能上的差异。
本篇博客的完整代码在我的github上：https://github.com/1399852153/Reinventing-the-wheel-for-learning（AQS模块）。
由于AQS无锁并发机制的复杂性，可能MyAQSV1在裁剪、改写jdk实现的过程中无意中引入了一些bug，如有错误，还请多多指教。
-
文件掩码什么意思？-icode9专业技术文章分享12-27
-
如何使用循环来处理多个订单的退款请求，代码怎么写？-icode9专业技术文章分享12-27
-
VSCode 在编辑时切换到另一个文件后再切回来如何保持在原来的位置？-icode9专业技术文章分享12-27
-
Sealos Devbox 基础教程：使用 Cursor 从零开发一个 One API 替代品 审核中12-27
-
TypeScript面试真题解析与实战指南12-27
-
TypeScript大厂面试真题详解与解析12-27
-
怎么使用nsenter命令进入容器？-icode9专业技术文章分享12-26
-
导入文件提示存在乱码，请确定使用的是UTF-8编码怎么解决？-icode9专业技术文章分享12-26
-
csv文件怎么设置编码？-icode9专业技术文章分享12-26
-
TypeScript基础知识详解12-25
-
安卓NDK 是什么？-icode9专业技术文章分享12-25
-
caddy 可以定义日志到 文件吗？-icode9专业技术文章分享12-25
-
wordfence如何设置密码规则？-icode9专业技术文章分享12-25
-
有哪些方法可以实现 DLL 文件路径的管理？-icode9专业技术文章分享12-25
-
错误信息 "At least one element in the source array could not be cast down to the destination array-icode9专业技术文章分享12-25
