【java基础】独占锁ReentrantLock对AQS实现的源码分析
Lock是一个接口，而synchronized是Java中的关键字，synchronized是基于jvm实现。Lock锁可以被中断，支持定时锁等。Lock的实现类，可重入锁ReentrantLock，我们有讲到其具体用法。而谈到ReentrantLock，不得不谈抽象类AbstractQueuedSynchronizer（AQS）。抽象的队列式的同步器，AQS定义了一套多线程访问共享资源的同步器框架，许多同步类实现都依赖于它，如常用的ReentrantLock、ThreadPoolExecutor。
2. AQS介绍
AQS是一个抽象类，主是是以继承的方式使用。AQS本身是没有实现任何同步接口的，它仅仅只是定义了同步状态的获取和释放的方法来供自定义的同步组件的使用。AQS抽象类包含如下几个关键字段：
/**同步队列的头节点，延迟初始化。他只能通过setHead()修改
* 注意：如果头节点存在，他的waitStatus保证不能是CANCELLED
*/
private transient volatile Node head;
/**同步队列的尾节点，延迟初始化。只能通过增加新节点的enq()修改和addWaiter()
*/
private transient volatile Node tail;
/**
* 同步状态，多线程进行抢夺的资源，独享锁只能有一个线程获取
*/
private volatile int state;
/**当前独享模式下的同步锁拥有者线程
*/
private transient Thread exclusiveOwnerThread;
AQS定义两种资源共享方式：Exclusive（独占，只有一个线程能执行，如ReentrantLock）和Share（共享，多个线程可同时执行，如Semaphore/CountDownLatch）。共享模式时只用 Sync Queue, 独占模式有时只用 Sync Queue, 但若涉及 Condition, 则还有 Condition Queue。在子类的 tryAcquire, tryAcquireShared 中实现公平与非公平的区分。
ReentrantLock内部实现了两种AQS，公平锁FairSync和非公平锁NonfairSync。
abstract static class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = -5179523762034025860L;
/**
* 线程获取锁，由子类实现，可以实现为公平的还是非公平的
*/
abstract void lock();
/**
* 非公平锁获取
*/
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();//获取当前AQS中同步资源state值
if (c == 0) {
if (compareAndSetState(0, acquires)) {
//如果当前线程修改成功，则获取锁，将当前线程保存在AQS的exclusiveOwnerThread，返回true
setExclusiveOwnerThread(current);
return true;
}
}
//如果当前线程已经获取过锁，则可重入，state值增加，不用重新获取锁
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;
}
//释放锁，将state值置0，将exclusiveOwnerThread置空
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;
}
//判断当前线程是否是同步锁的拥有着
protected final boolean isHeldExclusively() {
return getExclusiveOwnerThread() == Thread.currentThread();
}
//ReentrantLock进行条件等待时候创建的对象
final ConditionObject newCondition() {
return new ConditionObject();
}
//获取同步锁当前拥有着
final Thread getOwner() {
return getState() == 0 ? null : getExclusiveOwnerThread();
}
//获取当前线程对state的拥有数值。
final int getHoldCount() {
return isHeldExclusively() ? getState() : 0;
}
//判断是否有线程占用锁
final boolean isLocked() {
return getState() != 0;
}
}不同的自定义同步器争用共享资源的方式也不同。自定义同步器在实现时只需要实现共享资源state的获取与释放方式即可，至于具体线程等待队列的维护（如获取资源失败入队/唤醒出队等），AQS已经在顶层实现好了。
整个 AQS 分为以下几部分:
- Node 节点, 用于存放获取线程的节点, 存在于 Sync Queue, Condition Queue, 这些节点主要的区分在于 waitStatus 的值(下面会详细叙述)
- Condition Queue, 这个队列是用于独占模式中, 只有用到 Condition.awaitXX 时才会将 node加到 tail 上(PS: 在使用 Condition的前提是已经获取 Lock)
- Sync Queue, 独占 共享的模式中均会使用到的存放 Node 的 CLH queue(主要特点是, 队列中总有一个 dummy 节点, 后继节点获取锁的条件由前继节点决定, 前继节点在释放 lock 时会唤醒sleep中的后继节点)
- ConditionObject, 用于独占的模式, 主要是线程释放lock, 加入 Condition Queue, 并进行相应的 signal 操作。
- 独占的获取lock (acquire, release), 例如 ReentrantLock。
- 共享的获取lock (acquireShared, releaseShared), 例如 ReeantrantReadWriteLock, Semaphore, CountDownLatch
3. 内部类 Node
Node 节点是代表获取lock的线程, 存在于 Condition Queue, Sync Queue 里面， 而其主要就是 nextWaiter (标记共享还是独占),waitStatus 标记node的状态。
static final class Node {
/** 标识节点是否是 共享的节点(这样的节点只存在于 Sync Queue 里面) */
static final Node SHARED = new Node();
//独占模式
static final Node EXCLUSIVE = null;
/**
* CANCELLED 说明节点已经 取消获取 lock 了(一般是由于 interrupt 或 timeout 导致的)
* 很多时候是在 cancelAcquire 里面进行设置这个标识
*/
static final int CANCELLED = 1;
/**
* SIGNAL 标识当前节点的后继节点需要唤醒(PS: 这个通常是在 独占模式下使用, 在共享模式下有时用 PROPAGATE)
*/
static final int SIGNAL = -1;
//当前节点在 Condition Queue 里面
static final int CONDITION = -2;
/**
* 当前节点获取到 lock 或进行 release lock 时, 共享模式的最终状态是 PROPAGATE(PS: 有可能共享模式的节点变成 PROPAGATE 之前就被其后继节点抢占 head 节点, 而从Sync Queue中被踢出掉)
*/
static final int PROPAGATE = -3;
volatile int waitStatus;
/**
* 节点在 Sync Queue 里面时的前继节点(主要来进行 skip CANCELLED 的节点)
* 注意: 根据 addWaiter方法:
* 1. prev节点在队列里面, 则 prev != null 肯定成立
* 2. prev != null 成立, 不一定 node 就在 Sync Queue 里面
*/
volatile Node prev;
/**
* Node 在 Sync Queue 里面的后继节点, 主要是在release lock 时进行后继节点的唤醒
* 而后继节点在前继节点上打上 SIGNAL 标识, 来提醒他 release lock 时需要唤醒
*/
volatile Node next;
//获取 lock 的引用
volatile Thread thread;
/**
* 作用分成两种:
* 1. 在 Sync Queue 里面, nextWaiter用来判断节点是 共享模式, 还是独占模式
* 2. 在 Condition queue 里面, 节点主要是链接且后继节点 (Condition queue是一个单向的, 不支持并发的 list)
*/
Node nextWaiter;
// 当前节点是否是共享模式
final boolean isShared() {
return nextWaiter == SHARED;
}
// 获取 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 用于 Sync Queue 里面 addWaiter(Node.EXCLUSIVE)
Node(Thread thread, Node mode){ // Used by addWaiter
this.nextWaiter = mode;
this.thread = thread;
}
//初始化 Node 用于 Condition Queue 里面
//Node node = new Node(Thread.currentThread(), Node.CONDITION);
Node(Thread thread, int waitStatus){ // Used by Condition
this.waitStatus = waitStatus;
this.thread = thread;
}
}
waitStatus的状态变化:
- 线程还未入 Sync Queue 里面,先尝试获取一下锁(调用 tryAcquire 方法) （非公平，后来的线程有可能不用排队就可以获取锁），发现独占锁被其他人获取, 就入队列。如果前继节点是head节点就进行自旋并tryAcquire。
- 否则判断一下是否前继节点被标记为 SIGNAL, 若是的话 直接 block(block前会确保前继节点被标记为SIGNAL, 因为前继节点在进行释放锁时根据是否标记为 SIGNAL 来决定唤醒后继节点与否 <- 这是独占的情况下)
- 前继节点使用完lock, 进行释放, 因为自己被标记为 SIGNAL, 所以唤醒其后继节点
waitStatus 变化过程:
- 独占模式下: 0(初始) -> signal(被后继节点标记为release需要唤醒后继节点) -> 0 (等释放好lock, 会恢复到0)
- 独占模式 + 使用 Condition情况下: 0(初始) -> signal(被后继节点标记为release需要唤醒后继节点) -> 0 (等释放好lock, 会恢复到0)其上可能涉及 中断与超时, 只是多了一个 CANCELLED, 当节点变成 CANCELLED, 后就等着被清除。
- 共享模式下: 0(初始) -> PROPAGATE(获取 lock 或release lock 时) (获取 lock 时会调用 setHeadAndPropagate 来进行 传递式的唤醒后继节点, 直到碰到 独占模式的节点)
- 共享模式 + 独占模式下: 0(初始) -> signal(被后继节点标记为release需要唤醒后继节点) -> 0 (等释放好lock, 会恢复到0)
其上的这些状态变化主要在: doReleaseShared , shouldParkAfterFailedAcquire里面。
4. Sync Queue
AQS内部维护着一个FIFO的CLH队列，所以AQS并不支持基于优先级的同步策略。至于为何要选择CLH队列，主要在于CLH锁相对于MSC锁，他更加容易处理cancel和timeout，同时他具备进出队列快、无所、畅通无阻、检查是否有线程在等待也非常容易（head != tail,头尾指针不同）。当然相对于原始的CLH队列锁，ASQ采用的是一种变种的CLH队列锁：
- 原始CLH使用的locked自旋，而AQS的CLH则是在每个node里面使用一个状态字段来控制阻塞，而不是自旋。
- 为了可以处理timeout和cancel操作，每个node维护一个指向前驱的指针。如果一个node的前驱被cancel，这个node可以前向移动使用前驱的状态字段。
- head结点使用的是傀儡结点。
4.1 Sync Queue 节点入Queue方法
这里有个地方需要注意, 就是初始化 head, tail 的节点, 不一定是 head.next, 因为期间可能被其他的线程进行抢占了。将当前的线程封装成 Node 加入到 Sync Queue 里面
private Node addWaiter(Node mode){
Node node = new Node(Thread.currentThread(), mode); // 1. 封装 Node
Node pred = tail;
if(pred != null){ // 2. pred != null -> 队列中已经有节点, 直接 CAS 到尾节点
node.prev = pred; // 3. 先设置 Node.pre = pred (PS: 则当一个 node在Sync Queue里面时 node.prev 一定 != null(除 dummy node), 但是 node.prev != null 不能说明其在 Sync Queue 里面, 因为现在的CAS可能失败 )
if(compareAndSetTail(pred, node)){ // 4. CAS node 到 tail
pred.next = node; // 5. CAS 成功, 将 pred.next = node (PS: 说明 node.next != null -> 则 node 一定在 Sync Queue, 但若 node 在Sync Queue 里面不一定 node.next != null)
return node;
}
}
enq(node); // 6. 队列为空,或者CAS失败 调用 enq 入队列
return node;
}
/**
* 这个插入会检测head tail 的初始化, 必要的话会初始化一个 dummy 节点, 这个和 ConcurrentLinkedQueue 一样的
* 将节点 node 加入队列
* 这里有个注意点
* 情况:
* 1. 首先 queue是空的
* 2. 初始化一个 dummy 节点
* 3. 这时再在tail后面添加节点(这一步可能失败, 可能发生竞争被其他的线程抢占)
* 这里为什么要加入一个 dummy 节点呢?
* 这里的 Sync Queue 是CLH lock的一个变种, 线程节点 node 能否获取lock的判断通过其前继节点
* 而且这里在当前节点想获取lock时通常给前继节点 打上 signal 的标识(表示前继节点释放lock需要通知我来获取lock)
* 若这里不清楚的同学, 请先看看 CLH lock的资料 (这是理解 AQS 的基础)
*/
private Node enq(final Node node){
for(;;){
Node t = tail;
if(t == null){ // Must initialize // 1. 队列为空 初始化一个 dummy 节点 其实和 ConcurrentLinkedQueue 一样
if(compareAndSetHead(new Node())){ // 2. 初始化 head 与 tail (这个CAS成功后, head 就有值了, 详情将 Unsafe 操作)
tail = head;
}
}else{
node.prev = t; // 3. 先设置 Node.pre = pred (PS: 则当一个 node在Sync Queue里面时 node.prev 一定 != null, 但是 node.prev == null 说明其不在 Sync Queue 里面 )
if(compareAndSetTail(t, node)){ // 4. CAS node 到 tail
t.next = node; // 5. CAS 成功, 将 pred.next = node (PS: 说明 node.next != null -> 则 node 一定在 Sync Queue, 但若 node 在Sync Queue 里面不一定 node.next != null)
return t;
}
}
}
}
4.2 Sync Queue 节点出Queue方法
这里的出Queue的方法其实有两个： 新节点获取lock, 调用setHead抢占head, 并且剔除原head；节点因被中断或获取超时而进行 cancelled, 最后被剔除。
/**
* 设置 head 节点(在独占模式没有并发的可能, 当共享的模式有可能)
*/
private void setHead(Node node){
head = node;
node.thread = null; // 清除线程引用
node.prev = null; // 清除原来 head 的引用 <- 都是 help GC
}
// 清除因中断/超时而放弃获取lock的线程节点(此时节点在 Sync Queue 里面)
private void cancelAcquire(Node node) {
if (node == null)
return;
node.thread = null; // 1. 线程引用清空
Node pred = node.prev;
while (pred.waitStatus > 0) // 2. 若前继节点是 CANCELLED 的, 则也一并清除
node.prev = pred = pred.prev;
Node predNext = pred.next; // 3. 这里的 predNext也是需要清除的(只不过在清除时的 CAS 操作需要 它)
node.waitStatus = Node.CANCELLED; // 4. 标识节点需要清除
// If we are the tail, remove ourselves.
if (node == tail && compareAndSetTail(node, pred)) { // 5. 若需要清除额节点是尾节点, 则直接 CAS pred为尾节点
compareAndSetNext(pred, predNext, null); // 6. 删除节点predNext
} else {
int ws;
if (pred != head && //7. 不是头节点，因为头节点的后继节点要进行自旋随时等待获取锁
((ws = pred.waitStatus) == Node.SIGNAL || // 8. pred节点ws=SIGNAL
(ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) && // 9. 将 pred 标识为 SIGNAL
pred.thread != null) //10.不是已经释放锁的节点，因为pred在上面代码获取时，有可能该节点已经获取过锁然后释放锁成为废弃节点了
{
Node next = node.next;
if (next != null && next.waitStatus <= 0) // 11. next.waitStatus <= 0 表示 next 是个一个想要获取lock的节点
compareAndSetNext(pred, predNext, next);
} else {
unparkSuccessor(node); // 12.否则，对node的后继节点无法交代，必须唤醒后继系节点
}
node.next = node; // help GC
}
}
这里有几点注意：
- 只要获取锁成功成为head节点，就不会被取消，因为cancelAcquire方法永远不会执行。
- 在同步队列中的节点，要么在自旋获取锁，要么已经设置前继节点的waitStatus为SIGNAL,然后挂起睡眠了。
- 清除节点发生在shouldParkAfterFailedAcquire和cancelAcquire方法中。
5. 独占Lock
5.1 独占方式获取lock主要流程
- 调用 tryAcquire 尝试性的获取锁(一般都是由子类实现), 成功的话直接返回
- tryAcquire 调用获取失败, 将当前的线程封装成 Node 加入到 Sync Queue 里面(调用addWaiter), 等待获取 signal 信号
- 调用 acquireQueued 进行自旋的方式获取锁(有可能会 repeatedly blocking and unblocking)
- 根据acquireQueued的返回值判断在获取lock的过程中是否被中断, 若被中断, 则自己再中断一下(selfInterrupt), 若是响应中断的则直接抛出异常
5.2 独占方式获取lock主要分成3类
- acquire 不响应中断的获取lock, 这里的不响应中断指的是线程被中断后会被唤醒, 并且继续获取lock,在方法返回时, 根据刚才的获取过程是否被中断来决定是否要自己中断一下(方法 selfInterrupt)
- doAcquireInterruptibly 响应中断的获取 lock, 这里的响应中断, 指在线程获取 lock 过程中若被中断, 则直接抛出异常
- doAcquireNanos 响应中断及超时的获取 lock, 当线程被中断, 或获取超时, 则直接抛出异常, 获取失败
5.3 独占的获取lock 方法 acquire
acquire(int arg)：以独占模式获取对象，忽略中断。
public final void acquire(int arg){
if(!tryAcquire(arg)&&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) {
selfInterrupt();
}
}
- 调用 tryAcquire 尝试性的获取锁(一般都是由子类实现), 成功的话直接返回
- tryAcquire 调用获取失败, 将当前的线程封装成 Node 加入到 Sync Queue 里面(调用addWaiter), 等待获取 signal 信号
- 调用 acquireQueued 进行自旋的方式获取锁(有可能会 repeatedly blocking and unblocking)
- 根据acquireQueued的返回值判断在获取lock的过程中是否被中断, 若被中断, 则自己再中断一下(selfInterrupt)。
5.4 循环获取lock 方法 acquireQueued
final boolean acquireQueued(final Node node, int arg){
boolean failed = true;
try {
boolean interrupted = false;
for(;;){
final Node p = node.predecessor(); // 1. 获取当前节点的前继节点 (当一个n在 Sync Queue 里面, 并且没有获取 lock 的 node 的前继节点不可能是 null)
if(p == head && tryAcquire(arg)){ // 2. 判断前继节点是否是head节点(前继节点是head, 存在两种情况 (1) 前继节点现在占用 lock (2)前继节点是个空节点, 已经释放 lock, node 现在有机会获取 lock); 则再次调用 tryAcquire尝试获取一下
setHead(node); // 3. 获取 lock 成功, 直接设置 新head(原来的head可能就直接被回收)
p.next = null; // help GC // help gc
failed = false;
return interrupted; // 4. 返回在整个获取的过程中是否被中断过 ; 但这又有什么用呢? 若整个过程中被中断过, 则最后我在 自我中断一下 (selfInterrupt), 因为外面的函数可能需要知道整个过程是否被中断过
}
if(shouldParkAfterFailedAcquire(p, node) && // 5. 调用 shouldParkAfterFailedAcquire 判断是否需要中断(这里可能会一开始 返回 false, 但在此进去后直接返回 true(主要和前继节点的状态是否是 signal))
parkAndCheckInterrupt()){ // 6. 现在lock还是被其他线程占用 那就睡一会, 返回值判断是否这次线程的唤醒是被中断唤醒
interrupted = true;
}
}
}finally {
if(failed){ // 7. 在整个获取中出错
cancelAcquire(node); // 8. 清除 node 节点(清除的过程是先给 node 打上 CANCELLED标志, 然后再删除)
}
}
}
逻辑分析：
- 当前节点的前继节点是head节点时，先 tryAcquire获取一下锁, 成功的话设置新 head, 返回
- 第一步不成功, 检测是否需要sleep, 需要的话就sleep, 等待前继节点在释放lock时唤醒或通过中断来唤醒
- 整个过程可能需要blocking nonblocking 几次
ps:LockSupport.park()是可以通过中断来唤醒的。挂起的节点有可能会多次挂起和唤醒，比如:
(1)当中断发生，唤醒，发现还轮不到自己获取锁，而且前继节点waitStatus是SIGNAL，所以又放心的挂起睡眠。
(2)前继节点A取消，A在取消的过程发现前继节点是头节点，于是A唤醒后继节点B，然后就走了，此时B开始执行，然而发现并获取不到锁，因为头节点在占用，而且发现头节点WaitStatus是SIGNAL，于是又开始挂起睡眠。
这里注意一下5和6，比如节点A在5执行完成发现前继节点B的waitStatus是SIGNAL于是返回true,表明A可以放心的挂起睡眠了，但此时有可能cpu将执行权正好给了前继节点B里面的线程，并且前继节点线程完成，释放锁，发现自己的waitStatus是SIGNAL，于是唤醒A，但是A还没有执行parkAndCheckInterrupt挂起睡眠，当cpu重新分配时间片给A时，A执行parkAndCheckInterrupt挂起睡眠，这里有个现后顺序，就是B给A发个通行证，然后A挂起睡眠，但是发现这是A已经有通行证了，不需要挂起了，就继续开始执行线程。这里涉及到的知识点是LockSupport的park和unPark的执行顺序问题。
一个线程它有可能在别的线程unPark之前，或者之后，或者同时调用了park，那么因为park的特性，它可以不用担心自己的park的时序问题。
5.5 支持中断获取lock 方法 doAcquireInterruptibly
private void doAcquireInterruptibly(int arg) throws InterruptedException{
final Node node = addWaiter(Node.EXCLUSIVE); // 1. 将当前的线程封装成 Node 加入到 Sync Queue 里面
boolean failed = true;
try {
for(;;){
final Node p = node.predecessor(); // 2. 获取当前节点的前继节点 (当一个n在 Sync Queue 里面, 并且没有获取 lock 的 node 的前继节点不可能是 null)
if(p == head && tryAcquire(arg)){ // 3. 判断前继节点是否是head节点(前继节点是head, 存在两种情况 (1) 前继节点现在占用 lock (2)前继节点是个空节点, 已经释放 lock, node 现在有机会获取 lock); 则再次调用 tryAcquire尝试获取一下
setHead(node);
p.next = null; // help GC
failed = false;
return;
}
if(shouldParkAfterFailedAcquire(p, node) && // 4. 调用 shouldParkAfterFailedAcquire 判断是否需要中断(这里可能会一开始 返回 false, 但在此进去后直接返回 true(主要和前继节点的状态是否是 signal))
parkAndCheckInterrupt()){ // 5. 现在lock还是被其他线程占用 那就睡一会, 返回值判断是否这次线程的唤醒是被中断唤醒
throw new InterruptedException(); // 6. 线程此时唤醒是通过线程中断, 则直接抛异常
}
}
}finally {
if(failed){ // 7. 在整个获取中出错(比如线程中断)
cancelAcquire(node); // 8. 清除 node 节点(清除的过程是先给 node 打上 CANCELLED标志, 然后再删除)
}
}
}acquireInterruptibly(int arg)： 以独占模式获取对象，如果被中断则中止。
public final void acquireInterruptibly(int arg) throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (!tryAcquire(arg))
doAcquireInterruptibly(arg);
}通过先检查中断的状态，然后至少调用一次tryAcquire，返回成功。否则，线程在排队，不停地阻塞与唤醒，调用tryAcquire直到成功或者被中断。
5.6 超时&中断获取lock 方法
tryAcquireNanos(int arg, long nanosTimeout):独占且支持超时模式获取： 带有超时时间，如果经过超时时间则会退出。
private boolean doAcquireNanos(int arg, long nanosTimeout) throws InterruptedException{
if(nanosTimeout <= 0L){
return false;
}
final long deadline = System.nanoTime() + nanosTimeout; // 0. 计算截至时间
final Node node = addWaiter(Node.EXCLUSIVE); // 1. 将当前的线程封装成 Node 加入到 Sync Queue 里面
boolean failed = true;
try {
for(;;){
final Node p = node.predecessor(); // 2. 获取当前节点的前继节点 (当一个n在 Sync Queue 里面, 并且没有获取 lock 的 node 的前继节点不可能是 null)
if(p == head && tryAcquire(arg)){ // 3. 判断前继节点是否是head节点(前继节点是head, 存在两种情况 (1) 前继节点现在占用 lock (2)前继节点是个空节点, 已经释放 lock, node 现在有机会获取 lock); 则再次调用 tryAcquire尝试获取一下
setHead(node);
p.next = null; // help GC
failed = false;
return true;
}
nanosTimeout = deadline - System.nanoTime(); // 4. 计算还剩余的时间
if(nanosTimeout <= 0L){ // 5. 时间超时, 直接返回
return false;
}
if(shouldParkAfterFailedAcquire(p, node) && // 6. 调用 shouldParkAfterFailedAcquire 判断是否需要中断(这里可能会一开始 返回 false, 但在此进去后直接返回 true(主要和前继节点的状态是否是 signal))
nanosTimeout > spinForTimeoutThreshold){ // 7. 若没超时, 并且大于spinForTimeoutThreshold, 则线程 sleep(小于spinForTimeoutThreshold, 则直接自旋, 因为效率更高 调用 LockSupport 是需要开销的)
LockSupport.parkNanos(this, nanosTimeout);
}
if(Thread.interrupted()){ // 8. 线程此时唤醒是通过线程中断, 则直接抛异常
throw new InterruptedException();
}
}
}finally {
if(failed){ // 9. 在整个获取中出错(比如线程中断/超时)
cancelAcquire(node); // 10. 清除 node 节点(清除的过程是先给 node 打上 CANCELLED标志, 然后再删除)
}
}
}
尝试以独占模式获取，如果中断和超时则放弃。实现时先检查中断的状态，然后至少调用一次tryAcquire。
public final boolean tryAcquireNanos(int arg, long nanosTimeout) throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
return tryAcquire(arg)|| doAcquireNanos(arg, nanosTimeout);
}
5.7 释放lock方法
释放 lock 流程：
- 调用子类的 tryRelease 方法释放获取的资源
- 判断是否完全释放lock(这里有 lock 重复获取的情况)
- 判断是否有后继节点需要唤醒, 需要的话调用unparkSuccessor进行唤醒
public final boolean release(int arg){
if(tryRelease(arg)){ // 1. 调用子类, 若完全释放好, 则返回true(这里有lock重复获取)
Node h = head;
if(h != null && h.waitStatus != 0){ // 2. h.waitStatus !=0 其实就是 h.waitStatus < 0 后继节点需要唤醒
unparkSuccessor(h); // 3. 唤醒后继节点
}
return true;
}
return false;
}
/**
* 唤醒 node 的后继节点
* 这里有个注意点: 唤醒时会将当前node的标识归位为 0
* 等于当前节点标识位 的流转过程: 0(刚加入queue) -> signal (被后继节点要求在释放时需要唤醒) -> 0 (进行唤醒后继节点)
*/
private void unparkSuccessor(Node node) {
logger.info("unparkSuccessor node:" + node + Thread.currentThread().getName());
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0); // 1. 清除前继节点的标识
Node s = node.next;
logger.info("unparkSuccessor s:" + node + Thread.currentThread().getName());
if (s == null || s.waitStatus > 0) { // 2. 这里若在 Sync Queue 里面存在想要获取 lock 的节点,则一定需要唤醒一下(跳过取消的节点) （PS: s == null发生在共享模式的竞争释放资源）
s = null;
for (Node t = tail; t != null && t != node; t = t.prev)
if (t.waitStatus <= 0) // 3. 找到 queue 里面最前面想要获取 Lock 的节点
s = t;
}
logger.info("unparkSuccessor s:"+s);
if (s != null)
LockSupport.unpark(s.thread);
}
6. Condition Queue
Condition Queue 是一个并发不安全的, 只用于独占模式的队列(PS: 为什么是并发不安全的呢? 主要是在操作 Condition 时, 线程必需获取 独占的 lock, 所以不需要考虑并发的安全问题); 而当Node存在于 Condition Queue 里面, 则其只有 waitStatus, thread, nextWaiter 有值, 其他的都是null(其中的 waitStatus 只能是 CONDITION, 0(0 代表node进行转移到 Sync Queue里面, 或被中断/timeout)); 这里有个注意点, 就是当线程被中断或获取 lock 超时, 则一瞬间 node 会存在于 Condition Queue, Sync Queue 两个队列中.
节点 Node4, Node5, Node6, Node7 都是调用 Condition.awaitXX 方法加入 Condition Queue(PS: 加入后会将原来的 lock 释放)。
6.1 入队列方法 addConditionWaiter
将当前线程封装成一个 Node 节点放入到 Condition Queue 里面大家可以注意到, 下面对 Condition Queue 的操作都没考虑到 并发(Sync Queue 的队列是支持并发操作的), 这是为什么呢? 因为在进行操作 Condition 是当前的线程已经获取了AQS的独占锁, 所以不需要考虑并发的情况。
private Node addConditionWaiter(){
Node t = lastWaiter;
// Condition queue 的尾节点
// 尾节点已经Cancel, 直接进行清除,
/**
* 当Condition进行 awiat 超时或被中断时, Condition里面的节点是没有被删除掉的, 需要其 * 他await 在将线程加入 Condition Queue 时调用addConditionWaiter而进而删除, 或 await 操作差不多结束时, 调用 "node.nextWaiter != null" 进行判断而删除 (PS: 通过 signal 进行唤
* 醒时 node.nextWaiter 会被置空, 而中断和超时时不会)
*/
if(t != null && t.waitStatus != Node.CONDITION){
/**
* 调用 unlinkCancelledWaiters 对 "waitStatus != Node.CONDITION" 的节点进行 * 删除(在Condition里面的Node的waitStatus 要么是CONDITION(正常), 要么就是 0
* (signal/timeout/interrupt))
*/
unlinkCancelledWaiters();
t = lastWaiter;
}
//将线程封装成 node 准备放入 Condition Queue 里面
Node node = new Node(Thread.currentThread(), Node.CONDITION);
if(t == null){
//Condition Queue 是空的
firstWaiter = node;
} else {
// 追加到 queue 尾部
t.nextWaiter = node;
}
lastWaiter = node;
return node;
}
6.2 Condition的关键方法await()
await()逻辑：
- 如果当前线程中断，抛出InterruptedException
- 获取当前线程的state数值，然后通过release方法释放state,也就是释放锁。
- 挂起直到对应的signal()方法或者被中断。
- 唤醒后调用acquireQueued()去尝试获取锁，到这步就和线程刚进入同步队列去争夺锁步骤一样了。
- 注意，如果3中的中断唤醒发生在signal（）之前就throw InterruptedException,如果在之后就调用selfInterrupt（）标记线程中断。
public final void await() throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException(); //1.如果线程中断抛出InterruptedException
Node node = addConditionWaiter(); //2.调用addConditionWaiter将当前线程如等待队列
int savedState = fullyRelease(node); //3.释放当前线程占用的锁
int interruptMode = 0;
while (!isOnSyncQueue(node)) { //4.判断是否在sync队列上，如果节点WaitStatus=CONDTTION或者节点prev为null,那么节点一定不再sync队列，如果节点next不为空，那么一定在sync队列上，或者根sync队列节点逐一比较。
LockSupport.park(this); //5.挂起当前线程
/**************************************此处是await()方法的分割线到这里先看signal()方法在回过头看后面代码*******************************************/
///此时节点已经被signal()方法加入到同步队列中了，然后调用acquireQueued进行自旋或者挂起等待锁。
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break; //6. 如果等待过程发生中断，中断唤醒发生在signal（）之前就throw InterruptedException,如果在之后就调用selfInterrupt（）标记线程
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT; //7.调用acquireQueued进行自旋或者挂起等待锁
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters(); //8.删除waitStatus!=CONDITION的节点
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);//9.判断抛出InterruptedException还是调用selfInterrupt()
}6.3 删除Cancelled节点的方法 unlinkCancelledWaiters
当Node在Condition Queue 中, 若状态不是 CONDITION, 则一定是被中断或超时。在调用 addConditionWaiter 将线程放入 Condition Queue 里面时或 awiat 方法获取结束时 进行清理 Condition queue 里面的因 timeout/interrupt 而还存在的节点。这个删除操作比较巧妙, 其中引入了 trail 节点， 可以理解为traverse整个 Condition Queue 时遇到的最后一个有效的节点。
private void unlinkCancelledWaiters(){
Node t = firstWaiter;
Node trail = null;
while(t != null){
Node next = t.nextWaiter; // 1. 先初始化 next 节点
if(t.waitStatus != Node.CONDITION){ // 2. 节点不有效, 在Condition Queue 里面 Node.waitStatus 只有可能是 CONDITION 或是 0(timeout/interrupt引起的)
t.nextWaiter = null; // 3. Node.nextWaiter 置空
if(trail == null){ // 4. 一次都没有遇到有效的节点
firstWaiter = next; // 5. 将 next 赋值给 firstWaiter(此时 next 可能也是无效的, 这只是一个临时处理)
} else {
trail.nextWaiter = next; // 6. next 赋值给 trail.nextWaiter, 这一步其实就是删除节点 t
}
if(next == null){ // 7. next == null 说明 已经 traverse 完了 Condition Queue
lastWaiter = trail;
}
}else{
trail = t; // 8. 将有效节点赋值给 trail
}
t = next;
}
}
6.4 Condition的关键方法signal()
public final void signal() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node first = firstWaiter;
if (first != null)
doSignal(first);
}
//找到第一个waitStatus=CONDITION的节点，将此节点waitStatus=0，入sync队列
private void doSignal(Node first) {
do {
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
first.nextWaiter = null;
} while (!transferForSignal(first) &&
(first = firstWaiter) != null);
}
6.5 转移节点的方法 transferForSignal
transferForSignal只有在节点被正常唤醒才调用的正常转移的方法。
将Node 从Condition Queue 转移到 Sync Queue 里面在调用transferForSignal之前, 会 first.nextWaiter = null;而我们发现若节点是因为 timeout / interrupt 进行转移, 则不会进行这步操作; 两种情况的转移都会把 wautStatus 置为 0
final boolean transferForSignal(Node node){
/**
* If cannot change waitStatus, the node has been cancelled
*/
if(!compareAndSetWaitStatus(node, Node.CONDITION, 0)){ // 1. 若 node已经被其他线程调用signal()加入到sync队列,则失败。
return false;
}
Node p = enq(node); // 2. 加入 Sync Queue
int ws = p.waitStatus;
if(ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)){ // 3. 这里的 ws > 0 指Sync Queue 中node 的前继节点cancelled 了, 所以, 唤醒一下 node ; compareAndSetWaitStatus(p, ws, Node.SIGNAL)失败, 则说明 前继节点已经变成 SIGNAL 或 cancelled, 所以也要 唤醒
LockSupport.unpark(node.thread);
}
return true;
}
下面画图理解一下：
总结：
（1）多线程在遇到ReentrantLock时，只会有一个线程获取锁，并继续执行代码，其他线程就需要进入Sync队列进行排队，按照FIFO规则进出，但是ReentrantLock可以实现两种模式，公平锁和非公锁，公平锁就是严格按照Sync队列顺序依次获取锁，新进来的线程必须先入Sync队列，然后严格按照顺序获取锁。而非公平锁意思是，已经在Sync队列中的线程必须按照顺序依次获取锁，但是，刚进来的线程可以先获取锁，如果获取成功就不用入Sync队列，这就是不公平的体现。
（2）在Sync队列中的节点，要么在自旋获取锁，要么就是已经设置前继节点SIGNAL然后挂起睡眠了。挂起睡眠过程中可能是由于前继节点唤醒，也可能是中断或者超时唤醒。FairSync和NonfairSync重写了lock()和tryAquire()来实现公平和非公平获取锁。所以ReentrantLock.lock()是会忽略中断，lockInterruptibly()会响应中断，tryLock(long timeout, TimeUnit unit)是带超时的获取支持中断响应，这三个函数公平性由ReentrantLock实现了FairSync还是NonfairSync决定。但有一个函数特殊，tryLock()它是直接使用nonfairTryAcquire一次性尝试获取锁，是非公平的，获取到锁就返回true,否则就返回false。
（3）如果在ReentrantLock中使用了Condition,那么await()方法会释放占用的锁，并将当前线程加入到等待队列Condition Queue，挂起当前线程线程（有可能会跳过，因为刚把节点加入Condition Queue,就有另外线程使用signal()将这个节点转移到Sync Queue），知道对应的signal()方法唤醒，或者发生中断，如果唤醒是中断唤醒，就需要进行判断：
1.在signal()方法调用之前发生中断，就抛出InterruptedException，响应中断
2.如果是在signal()方法调用之后发生中断，就调用selfInterrupt()进行中断标记，忽略中断。
(4)ReentrantLock的tryLock(long, TimeUnit):独占且支持超时模式获取： 带有超时时间，如果经过超时时间则会退出，放弃获取锁。Condition的await(long, TimeUnit)超时会从条件队列转移到同步等待队列去获取锁。
下面是Condition接口定义的方法:
- void await() throws InterruptedException
当前线程进入等待状态，直到被通知（signal）或者被中断时，当前线程进入运行状态，从await()返回；
- void awaitUninterruptibly()
当前线程进入等待状态，直到被通知，对中断不做响应；
- long awaitNanos(long nanosTimeout) throws InterruptedException
在接口1的返回条件基础上增加了超时响应，返回值表示当前剩余的时间，如果在nanosTimeout之前被唤醒，返回值 = nanosTimeout - 实际消耗的时间，返回值 <= 0表示超时；
- boolean await(long time, TimeUnit unit) throws InterruptedException
同样是在接口1的返回条件基础上增加了超时响应，与接口3不同的是：可以自定义超时时间单位；返回值返回true/false，在time之前被唤醒，返回true，超时返回false。
- boolean awaitUntil(Date deadline) throws InterruptedException
当前线程进入等待状态直到将来的指定时间被通知，如果没有到指定时间被通知返回true，否则，到达指定时间，返回false；
参考文献：