AQSReentrantLock

一、AQS

在 Lock 中&＃xff0c;用到了一个同步队列 AQS&＃xff0c;全称 AbstractQueuedSynchronizer&＃xff0c;它是一个同步工具&＃xff0c;也是 Lock 用来实现线程同步的核心组件。

1.AQS 的两种功能

独占和共享。

• 独占锁&＃xff1a;每次只能有一个线程持有锁&＃xff0c;ReentrantLock 就是以独占方式实现的互斥锁。
• 共享锁 &＃xff0c;允许多个线程同时获取锁 &＃xff0c;并发访问共享资源 &＃xff0c;比如ReentrantReadWriteLock

2.AQS的实现

AQS 队列内部维护的是一个 FIFO 的双向链表&＃xff0c;这种结构的特点是每个数据结构都有两个指针&＃xff0c;分别指向直接的后继节点和直接前驱节点。所以双向链表可以从任意一个节点开始很方便的访问前驱和后继。
每个 Node 其实是由线程封装&＃xff0c;当线程争抢锁失败后会封装成 Node 加入到 AQS 队列中去&＃xff1b; 当获取锁的线程释放锁以后&＃xff0c;会从队列中唤醒一个阻塞的节点(线程)。

Node组成&＃xff1a;

static final class Node {
// 排他锁的标识
static final Node EXCLUSIVE &＃61; null;
// 如果带有这个标识&＃xff0c;证明是失效了
static final int CANCELLED &＃61; 1;
// 具有这个标识&＃xff0c;说明后继节点需要被唤醒
static final int SIGNAL &＃61; -1;
// Node对象存储标识的地方
volatile int waitStatus;
// 指向上一个节点
volatile Node prev;
// 指向下一个节点
volatile Node next;
// 当前Node绑定的线程
volatile Thread thread;
// 存储在Condition队列中的后继节点
Node nextWaiter;
// 返回前驱节点&＃xff0c;如果前驱节点为null&＃xff0c;抛出NPE
final Node predecessor() throws NullPointerException {
Node p &＃61; prev;
if (p &＃61;&＃61; null)
throw new NullPointerException();
else
return p;
}
// 将线程构造成一个Node&＃xff0c;添加到等待队列
Node(Thread thread, Node mode) { // Used by addWaiter
this.nextWaiter &＃61; mode;
this.thread &＃61; thread;
}
// 在Condition队列中使用
Node(Thread thread, int waitStatus) { // Used by Condition
this.waitStatus &＃61; waitStatus;
this.thread &＃61; thread;
}
}


3.AQS添加线程

• 
  
  1. 新的线程封装成 Node 节点追加到同步队列中&＃xff0c;设置 prev 节点以及修改当前节点的前置节点的 next 节点指向自己
• 
  
  2. 通过 CAS 将 tail 重新指向新的尾部节点

4.AQS释放锁

head 节点表示获取锁成功的节点&＃xff0c;当头结点在释放同步状态时&＃xff0c;会唤醒后继节点&＃xff0c;如果后继节点获得锁成功&＃xff0c;会把自己设置为头结点&＃xff1a;

• 
  
  1. 修改 head 节点指向下一个获得锁的节点
• 
  
  2. 新的获得锁的节点&＃xff0c;将 prev 的指针指向 null

设置 head 节点不需要用 CAS&＃xff0c;原因是设置 head 节点是由获得锁的线程来完成的&＃xff0c;而同步锁只能由一个线程获得&＃xff0c;所以不需要 CAS 保证&＃xff0c;只需要把 head 节点设置为原首节点的后继节点&＃xff0c;并且断开原 head 节点的 next 引用即可。

二、CAS

1.CAS 的实现原理

protected final boolean compareAndSetState(int expect, int update) {
// See below for intrinsics setup to support this
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}


通过 cas 乐观锁的方式来做比较并替换&＃xff0c;如果当前内存中的state 的值和预期值 expect 相等&＃xff0c;则替换为 update。更新成功返回 true&＃xff0c;否则返回 false。这个操作是原子的&＃xff0c;不会出现线程安全问题。

2.state属性

private volatile int state;
state 是 AQS 中的一个属性&＃xff0c;它在不同的实现中所表达的含义不一样&＃xff0c; 对于重入
锁的实现来说&＃xff0c;表示一个同步状态。它有两个含义的表示

• 当 state&＃61;0 时&＃xff0c;表示无锁状态
• 当 state>0 时&＃xff0c;表示已经有线程获得了锁&＃xff0c;也就是 state&＃61;1&＃xff0c;但是因为ReentrantLock 允许重入&＃xff0c;所以同一个线程多次获得同步锁的时候&＃xff0c; state 会递增&＃xff0c;比如重入 5 次&＃xff0c;那么 state&＃61;5。 而在释放锁的时候&＃xff0c;同样需要释放 5 次直到 state&＃61;0其他线程才有资格获得锁

3.Unsafe类

Unsafe 类是在 sun.misc 包下&＃xff0c;不属于 Java 标准。但是很多 Java 的基础类库,包括一些被广泛使用的高性能开发库都是基于 Unsafe 类开发的&＃xff0c;比如 Netty、
Hadoop、 Kafka 等&＃xff1b;

Unsafe 可认为是 Java 中留下的后门&＃xff0c;提供了一些低层次操作&＃xff0c;如直接内存访问、线程的挂起和恢复、 CAS、线程同步、内存屏障

而 CAS 就是 Unsafe 类中提供的一个原子操作&＃xff1a;
public final native boolean compareAndSwapInt(Object obj, long stateOffset, int expect, int update);

• obj&＃xff1a;需要改变的对象
• stateOffset&＃xff1a;偏移量(即之前求出来的 headOffset 的值)
• expect&＃xff1a;期待的值
• update&＃xff1a;更新后的值

整个方法的作用是如果当前时刻的值等于预期值 expect 相等&＃xff0c;则更新为新的期望值 update&＃xff0c;如果更新成功&＃xff0c;则返回 true&＃xff0c;否则返回false&＃xff1b;

stateOffset&＃xff1a;
一个 Java 对象可以看成是一段内存&＃xff0c;每个字段都得按照一定的顺序放在这段内存里&＃xff0c;通过这个方法可以准确地告诉你某个字段相对于对象的起始内存地址的字节偏移。用于在后面的 compareAndSwapInt 中&＃xff0c;去根据偏移量找到对象在内存中的具体位置
所以 stateOffset 表示 state 这个字段在 AQS 类的内存中相对于该类首地址的偏移量

compareAndSwapInt
unsafe.cpp 文件中compareAndSwarpInt 的实现&＃xff1a;

UNSAFE_ENTRY(jboolean, Unsafe_CompareAndSwapInt(JNIEnv *env, jobject unsafe, jobject obj, jlong offset,jint e, jint x))
UnsafeWrapper("Unsafe_CompareAndSwapInt");
oop p &＃61; JNIHandles::resolve(obj); //将 Java 对象解析成 JVM 的 oop&＃xff08;普通对象指针&＃xff09;
jint* addr &＃61; (jint *) index_oop_from_field_offset_long(p, offset); //根据对象 p 和地址偏移量找到地址
return (jint)(Atomic::cmpxchg(x, addr, e)) &＃61;&＃61; e; //基于 cas 比较并替换&＃xff0c; x 表示需要更新的值&＃xff0c; addr 表示 state在内存中的地址&＃xff0c; e 表示预期值
UNSAFE_END


三、ReentrantLock

ReentrantLock时序图&＃xff1a;

1.lock()

public void lock() {
// sync分为了公平和非公平
sync.lock();
}


sync是一个抽象的静态内部类&＃xff0c;它继承了 AQS 来实现重入锁的逻辑。
AQS 是一个同步队列&＃xff0c;它能够实现线程的阻塞以及唤醒&＃xff0c; 但它并不具备业务功能&＃xff0c; 所以在不同的同步场景中&＃xff0c;会继承 AQS 来实现对应场景的功能。

Sync 有两个具体的实现类&＃xff1a;

• NofairSync&＃xff1a;表示可以存在抢占锁的功能&＃xff0c;也就是说不管当前队列上是否存在其他线程等待&＃xff0c;新线程都有机会抢占锁
• FailSync: 表示所有线程严格按照 FIFO 来获取锁

NonfairSync#lock()&＃xff1a;

final void lock() {
// 通过CAS的方式尝试将state从0修改为1&＃xff0c;如果返回true&＃xff0c;代表修改成功&＃xff0c;如果修改失败&＃xff0c;返回false
if (compareAndSetState(0, 1))
// 将一个属性设置为当前线程&＃xff0c;这个属性是AQS的父类提供的
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}


1. 非公平锁和公平锁最大的区别在于&＃xff0c;非公平锁中抢占锁的逻辑是&＃xff0c;不管有没有线程排队&＃xff0c;直接cas去抢占
2. CAS 成功&＃xff0c;就表示成功获得了锁
3. CAS 失败&＃xff0c;调用 acquire(1)走锁竞争逻辑

2.AQS#acquire()

acquire 是 AQS 中的方法&＃xff0c;如果 CAS 操作未能成功&＃xff0c;说明 state 已经不为 0&＃xff0c;此时 acquire(1)操作

public final void acquire(int arg) {
// tryAcquire再次尝试获取锁资源&＃xff0c;如果尝试成功&＃xff0c;返回true
if (!tryAcquire(arg) &&
// 获取锁资源失败后&＃xff0c;需要将当前线程封装成一个Node&＃xff0c;追加到AQS的队列中
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
// 线程中断
selfInterrupt();
}


• 通过 tryAcquire 尝试获取独占锁&＃xff0c;如果成功返回 true&＃xff0c;失败返回 false
• 如果 tryAcquire 失败&＃xff0c;则会通过 addWaiter 方法将当前线程封装成 Node 添加到 AQS 队列尾部
• acquireQueued&＃xff0c;将 Node 作为参数&＃xff0c;通过自旋去尝试获取锁

3.NonfairSync#tryAcquire()

方法的作用是尝试获取锁&＃xff0c;如果成功返回 true&＃xff0c;不成功返回 false。
它是重写 AQS 类中的 tryAcquire 方法。

protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
final boolean nonfairTryAcquire(int acquires) {
// 获取当前线程
final Thread current &＃61; Thread.currentThread();
// 获取AQS的state的值
int c &＃61; getState();
// 如果state为0&＃xff0c;表示无锁状态&＃xff0c;尝试再次获取锁资源
if (c &＃61;&＃61; 0) {
// CAS尝试修改state&＃xff0c;从0-1&＃xff0c;如果成功&＃xff0c;设置ExclusiveOwnerThread属性为当前线程
if (compareAndSetState(0, acquires)) {
// 保存当前获得锁的线程,下次再来的时候不要再尝试竞争锁
setExclusiveOwnerThread(current);
return true;
}
}
// 当前占有锁资源的线程是否是当前线程
else if (current &＃61;&＃61; getExclusiveOwnerThread()) {
// 将state &＃43; 1
int nextc &＃61; c &＃43; acquires;
// 如果加1后&＃xff0c;小于0&＃xff0c;超所锁可重入的最大值&＃xff0c;抛出Error
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
// 没问题&＃xff0c;就重新对state进行复制
setState(nextc);
// 锁重入成功
return true;
}
return false;
}


4.AQS#addWaiter()

当 tryAcquire 方法获取锁失败以后&＃xff0c;则会先调用 addWaiter 将当前线程封装成Node.
入参 mode 表示当前节点的状态&＃xff0c;传递的参数是 Node.EXCLUSIVE&＃xff0c;表示独占状态。意味着重入锁用到了 AQS 的独占锁功能

• 1.将当前线程封装成 Node
• 2.当前链表中的 tail 节点是否为空&＃xff0c;如果不为空&＃xff0c;则通过 cas 操作把当前线程的node 添加到 AQS 队列
• 3.如果为空或者 cas 失败&＃xff0c;调用 enq 将节点添加到 AQS 队列

// 说明前面获取锁资源失败&＃xff0c;放到队列中等待
private Node addWaiter(Node mode) {
// 创建Node类&＃xff0c;并且设置thread为当前线程&＃xff0c;设置为排它锁
Node node &＃61; new Node(Thread.currentThread(), mode);
// 获取AQS中队列的尾部节点&＃xff0c;默认是 null
Node pred &＃61; tail;
// 如果tail !&＃61; null&＃xff0c;说明队列中存在节点
if (pred !&＃61; null) {
// 把当前线程的 Node 的 prev 指向 tail
node.prev &＃61; pred;
// 通过 cas 把 node加入到 AQS 队列&＃xff0c;也就是设置为 tail
if (compareAndSetTail(pred, node)) {
// 设置成功以后&＃xff0c;把原 tail 节点的 next指向当前 node
pred.next &＃61; node;
return node;
}
}
// tail&＃61;null,把 node 添加到同步队列
enq(node);
return node;
}
// enq &＃xff1a;通过自旋操作把当前节点加入到队列中
// 队列没有节点&＃xff0c;我是第一个&＃xff0c; 如果前面CAS失败&＃xff0c;也会进到这个位置重新往队尾进入。
private Node enq(final Node node) {
// 死循环
for (;;) {
// 重新获取当前的tail节点为t
Node t &＃61; tail;
if (t &＃61;&＃61; null) {
// 队列没有节点, 我是第一个&＃xff0c;没头没尾&＃xff0c;都是空
if (compareAndSetHead(new Node())) // 初始化一个Node作为head&＃xff0c;而这个head没有意义。
// 将头尾都指向了这个初始化的Node
tail &＃61; head;
} else {
// 有节点&＃xff0c;往队尾入
// 当前节点的上一个指向tail
node.prev &＃61; t;
// 基于CAS的方式&＃xff0c;将tail节点设置为当前节点
if (compareAndSetTail(t, node)) {
// 将之前的为节点的next&＃xff0c;设置为当前节点
t.next &＃61; node;
return t;
}
}
}
}


5.AQS#acquireQueued()

通过 addWaiter 方法把线程添加到链表后&＃xff0c; 会接着把 Node 作为参数传递给acquireQueued 方法&＃xff0c;去竞争锁&＃xff1a;

• 1.获取当前节点的 prev 节点
• 2.如果 prev 节点为 head 节点&＃xff0c;那么它就有资格去争抢锁&＃xff0c;调用 tryAcquire 抢占锁
• 3.抢占锁成功以后&＃xff0c;把获得锁的节点设置为 head&＃xff0c;并且移除原来的初始化 head节点
• 4.如果获得锁失败&＃xff0c;则根据 waitStatus 决定是否需要挂起线程
• 5.通过 cancelAcquire 取消获得锁的操作

// 已经将node加入到了双向队列中&＃xff0c;然后执行当前方法
final boolean acquireQueued(final Node node, int arg) {
// 标识
boolean failed &＃61; true;
try {
// 标识
boolean interrupted &＃61; false;
for (;;) {
// 获取当前节点的上一个节点p
final Node p &＃61; node.predecessor();
// 如果p是头&＃xff0c;说明有资格去争抢锁&＃xff0c;尝试获取锁资源&＃xff08;state从0-1&＃xff0c;锁重入操作&＃xff09;&＃xff0c;成功返回true&＃xff0c;失败返回false
if (p &＃61;&＃61; head && tryAcquire(arg)) {
// 获取锁成功&＃xff0c;设置head节点为当前节点&＃xff0c;将thread&＃xff0c;prev设置为null&＃xff0c;因为拿到锁资源了 &＃xff1b;
setHead(node);
p.next &＃61; null; // 把原 head 节点从链表中移除&＃xff0c;帮助GC回收
failed &＃61; false; // 将标识修改为false
return interrupted; // 返回interrupted
}
// 保证上一个节点是-1&＃xff0c;才会返回true&＃xff0c;才会将线程阻塞&＃xff0c;等待唤醒获取锁资源
if (shouldParkAfterFailedAcquire(p, node) &&
// 基于Unsafe类的park方法&＃xff0c;挂起线程
parkAndCheckInterrupt(); // 针对fail属性&＃xff0c;这里是唯一可能出现异常的地方&＃xff0c;JVM内部出现问题时&＃xff0c;可以这么理解&＃xff0c;fianlly代码块中的内容&＃xff0c;执行的几率约等于0
interrupted &＃61; true; // 返回当前线程在等待过程中有没有中断过
}
} finally {
if (failed)
cancelAcquire(node);
}
}


shouldParkAfterFailedAcquire
如果 ThreadA 的锁还没有释放的情况下&＃xff0c; ThreadB 和 ThreadC 来争抢锁肯定是会失败&＃xff0c;那么失败以后会调用 shouldParkAfterFailedAcquire 方法
Node 有 5 中状态

• CANCELLED&＃xff08;1&＃xff09; &＃xff1a;在同步队列中等待的线程等待超时或被中断&＃xff0c;需要从同步队列中取消该 Node 的结点, 其结点的 waitStatus 为 CANCELLED&＃xff0c;即结束状态&＃xff0c;进入该状态后的结点将不会再变化
• SIGNAL&＃xff08;-1&＃xff09; &＃xff1a;只要前置节点释放锁&＃xff0c;就会通知标识为 SIGNAL 状态的后续节点的线程
• CONDITION&＃xff08;-2&＃xff09;
• PROPAGATE(-3)&＃xff1a;享模式下&＃xff0c; PROPAGATE 状态的线程处于可运行状态
• 默认状态&＃xff08;0&＃xff09;
  通过 Node 的状态来判断&＃xff0c; ThreadA 竞争锁失败以后是否应该被挂起。

1. 如果 ThreadA 的 pred 节点状态为 SIGNAL&＃xff0c;那就表示可以放心挂起当前线程
2. 通过循环扫描链表把 CANCELLED 状态的节点移除
3. 修改 pred 节点的状态为 SIGNAL,返回 false.
   返回 false 时&＃xff0c;也就是不需要挂起&＃xff0c;返回 true&＃xff0c;则需要调用parkAndCheckInterrupt
   挂起当前线程

// node是当前节点&＃xff0c;pred是上一个节点
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
// 获取上一个节点的状态
int ws &＃61; pred.waitStatus;
// 如果上一个节点状态为SIGNAL&＃xff0c;意味着只需要等待其他前置节点的线程被释放
if (ws &＃61;&＃61; Node.SIGNAL)
return true; // 返回 true&＃xff0c;意味着可以直接放心的挂起了
// ws 大于 0&＃xff0c;意味着 prev 节点取消了排队&＃xff0c;直接移除这个节点
if (ws > 0) {
do {
// 将当前节点的prev指针指向了上一个的上一个
node.prev &＃61; pred &＃61; pred.prev;
} while (pred.waitStatus > 0); // 一直找到小于等于0的,从双向列表中移除 CANCELLED 的节点
// 将重新标识好的最近的有效节点的next
pred.next &＃61; node;
} else {
// 小于等于0&＃xff0c;不等于-1&＃xff0c;将上一个有效节点状态修改为-1
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}


parkAndCheckInterrupt:

private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}


使用 LockSupport.park 挂起当前线程变成 WATING 状态
Thread.interrupted&＃xff0c;返回当前线程是否被其他线程触发过中断请求&＃xff0c;也就是thread.interrupt(); 如果有触发过中断请求&＃xff0c;那么这个方法会返回当前的中断标识true&＃xff0c;并且对中断标识进行复位标识已经响应过了中断请求。 如果返回 true&＃xff0c;意味着在 acquire 方法中会执行 selfInterrupt()。

selfInterrupt:

static void selfInterrupt() {
Thread.currentThread().interrupt();
}


标识如果当前线程在 acquireQueued 中被中断过&＃xff0c;则需要产生一个中断请求&＃xff0c;原因是线程在调用 acquireQueued 方法的时候是不会响应中断请求的。

cancelAcquire&＃xff1a;

// cancelAcquire方法
private void cancelAcquire(Node node) {
// 如果当前节点为null&＃xff0c;结束&＃xff0c;健壮性判断
if (node &＃61;&＃61; null)
return;
// node不为null的前提下执行
// 将当前node的线程置位null &＃xff0c; 竞争锁资源跟我没有关系了&＃xff0c;
node.thread &＃61; null;
// 获取当前节点的前驱节点
Node pred &＃61; node.prev;
// 前驱节点的状态 > 0
while (pred.waitStatus > 0)
// 找到前驱中最近的非失效节点
node.prev &＃61; pred &＃61; pred.prev;
// 将第一个不是失效节点的后继节点声明出来
Node predNext &＃61; pred.next;
// 将当前节点置位失效节点。给别的Node看的。
node.waitStatus &＃61; Node.CANCELLED;
// 如果当前节点是尾节点&＃xff0c;将尾节点设置为最近的有效节点&＃xff08;如果当前节点为尾节点的操作&＃xff09;
if (node &＃61;&＃61; tail && compareAndSetTail(node, pred)) {
// 用CAS方式将尾节点的next设置null
compareAndSetNext(pred, predNext, null);
} else {
int ws;
// 中间节点操作
// 如果上一个节点不是头节点
if (pred !&＃61; head &&
获取上一届点状态&＃xff0c;是不是有效
((ws &＃61; pred.waitStatus) &＃61;&＃61; Node.SIGNAL || // pred需要唤醒后继节点的
(ws <&＃61; 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
pred.thread !&＃61; null) {
Node next &＃61; node.next;
if (next !&＃61; null && next.waitStatus <&＃61; 0)
compareAndSetNext(pred, predNext, next); // 尝试将pred的前驱节点的next指向当前节点的next&＃xff08;必须是有效的next节点&＃xff09;
} else {
// 头结点&＃xff0c;唤醒后继节点
unparkSuccessor(node);
}
node.next &＃61; node; // help GC
}
}