目录
- 使用
- 可重入
- 可打断
- 锁超时
- 公平锁
- 条件变量
- 原理
- 非公平锁实现原理
- 源码
- 流程
- 锁重入原理
- 可打断原理与不可打断原理
- 公平锁原理
- 条件变量原理
- await流程
- signal流程
使用
ReentrantLock是可冲入锁，与 synchronized 一样，都支持可重入。但是相对于 synchronized 它具备如下特点
- 可中断
- 可以设置超时时间
- 可以设置为公平锁
- 支持多个条件变量
ReentrantLock实现了Lock接口。
基本语法
// 获取锁
reentrantLock.lock();
try {
// 临界区
} finally {
// 释放锁
reentrantLock.unlock();
}
注意：锁【lock.lock】必须紧跟try代码块，且unlock要放到finally第一行。
下面来一一看下他的几个特点
可重入
可重入是指同一个线程如果首次获得了这把锁，那么因为它是这把锁的拥有者，因此有权利再次获取这把锁。如果是不可重入锁，那么第二次获得锁时，自己也会被锁挡住。
示例：
package up.cys.chapter03;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.locks.ReentrantLock;
@Slf4j
public class ReentrantLockTest01 {
static ReentrantLock lock = new ReentrantLock();
public static void main(String[] args) {
method1();
}
public static void method1() {
lock.lock();
try {
log.info("execute method1");
method2();
} finally {
lock.unlock();
}
}
public static void method2() {
lock.lock();
try {
log.info("execute method2");
method3();
} finally {
lock.unlock();
}
}
public static void method3() {
lock.lock();
try {
log.info("execute method3");
} finally {
lock.unlock();
}
}
}
输出：
2023-06-03 15:59:23,694 - 0 INFO [main] up.cys.chapter03.ReentrantLockTest01:23 - execute method1
2023-06-03 15:59:23,701 - 7 INFO [main] up.cys.chapter03.ReentrantLockTest01:32 - execute method2
2023-06-03 15:59:23,701 - 7 INFO [main] up.cys.chapter03.ReentrantLockTest01:41 - execute method3
可打断
可打断的意思是，再获取锁的过程中，可以打断，不再去获取锁。
获取锁时需要使用lock.lockInterruptibly()代替lock.lock()。
示例：
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.locks.ReentrantLock;
@Slf4j
public class ReentrantLockTest02 {
static ReentrantLock lock = new ReentrantLock();
public static void main(String[] args) {
Thread t1 = new Thread(() -> {
log.info("子线程启动...");
try {
lock.lockInterruptibly();
} catch (InterruptedException e) {
e.printStackTrace();
log.info("等锁的过程中被打断");
return;
}
try {
log.info("子线程获得了锁");
} finally {
lock.unlock();
}
}, "t1");
lock.lock();
try {
log.info("主线程获得了锁");
t1.start();
Thread.sleep(1000);
t1.interrupt();
log.info("主线程执行打断");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
输出如下：
2023-06-03 17:10:36,817 - 0 INFO [main] up.cys.chapter03.ReentrantLockTest02:35 - 主线程获得了锁
2023-06-03 17:10:36,823 - 6 INFO [t1] up.cys.chapter03.ReentrantLockTest02:18 - 子线程启动...
2023-06-03 17:10:37,830 - 1013 INFO [main] up.cys.chapter03.ReentrantLockTest02:39 - 主线程执行打断
2023-06-03 17:10:37,833 - 1016 INFO [t1] up.cys.chapter03.ReentrantLockTest02:23 - 等锁的过程中被打断
java.lang.InterruptedException
at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireInterruptibly(AbstractQueuedSynchronizer.java:944)
at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireInterruptibly(AbstractQueuedSynchronizer.java:1263)
at java.base/java.util.concurrent.locks.ReentrantLock.lockInterruptibly(ReentrantLock.java:317)
at up.cys.chapter03.ReentrantLockTest02.lambda$main$0(ReentrantLockTest02.java:20)
at java.base/java.lang.Thread.run(Thread.java:834)
注意如果是不可中断模式，那么即使使用了 interrupt 也不会让等待中断。
锁超时
获取锁时需要使用lock.tryLock()代替lock.lock()，意思是尝试获取锁，如果获取不到，则立即放弃。
还可以使用带参数的方法lock.tryLock(long timeout, TimeUnit unit)来设置尝试获取锁等待的时间。
示例：
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
@Slf4j
public class ReentrantLockTest03 {
public static void main(String[] args) {
ReentrantLock lock = new ReentrantLock();
Thread t1 = new Thread(() -> {
log.info("启动...");
try {
if (!lock.tryLock(1, TimeUnit.SECONDS)) {
log.info("获取等待 1s 后失败，返回");
return;
}
} catch (InterruptedException e) {
e.printStackTrace();
}
try {
log.info("获得了锁");
} finally {
lock.unlock();
}
}, "t1");
lock.lock();
try {
log.info("获得了锁");
t1.start();
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
输出如下：
2023-06-03 17:15:54,031 - 0 INFO [main] up.cys.chapter03.ReentrantLockTest03:38 - 获得了锁
2023-06-03 17:15:54,040 - 9 INFO [t1] up.cys.chapter03.ReentrantLockTest03:20 - 启动...
2023-06-03 17:15:55,048 - 1017 INFO [t1] up.cys.chapter03.ReentrantLockTest03:23 - 获取等待 1s 后失败，返回
公平锁
公平锁是指多个线程同时尝试获取同一把锁时，获取锁的顺序按照线程达到的顺序。对于非公平锁，则允许线程“插队”。
synchronized是非公平锁，而ReentrantLock的默认实现是非公平锁，但是也可以设置为公平锁。
ReentrantLock 默认是不公平的，即有些线程可能一直获取不到锁，出现饥饿。
改为公平锁：
ReentrantLock lock = new ReentrantLock(true);
公平锁会有个队列维护等待线程。公平锁一般没有必要，会降低并发度。
条件变量
synchronized 中也有条件变量，就是我们讲原理时那个 waitSet ，当条件不满足时进入 waitSet 等待。ReentrantLock 的条件变量比 synchronized 强大之处在于，它是支持多个条件变量的。synchronized 是那些不满足条件的线程都在一个条件变量等消息，而 ReentrantLock 支持多个条件变量t，唤醒时也只唤醒自己条件变量下等待的线程。
使用要点：
- await 前需要获得锁
- await 执行后，会释放锁，进入 conditionObject 等待
- await 的线程被唤醒（或打断、或超时）取重新竞争 lock 锁
- 竞争 lock 锁成功后，从 await 后继续执行
示例：
package up.cys.chapter03;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
@Slf4j
public class ReentrantLockTest03 {
static ReentrantLock lock = new ReentrantLock();
// 条件变量1：等待送烟
static Condition waitCigaretteQueue = lock.newCondition();
// 条件变量2：等待早餐
static Condition waitbreakfastQueue = lock.newCondition();
static volatile boolean hasCigrette = false;
static volatile boolean hasBreakfast = false;
public static void main(String[] args) {
// 线程1：等到烟才继续执行
new Thread(() -> {
lock.lock();
try {
log.info("线程1等待烟");
while (!hasCigrette) {
try {
waitCigaretteQueue.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
log.info("线程1等到了它的烟");
} finally {
lock.unlock();
}
}).start();
// 线程2：等到外卖才继续执行
new Thread(() -> {
lock.lock();
try {
log.info("线程2等早餐");
while (!hasBreakfast) {
try {
waitbreakfastQueue.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
log.info("线程2等到了它的早餐");
} finally {
lock.unlock();
}
}).start();
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 开始送早餐
sendBreakfast();
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 开始送烟
sendCigarette();
}
private static void sendCigarette() {
lock.lock();
try {
log.info("送烟来了");
hasCigrette = true;
waitCigaretteQueue.signal();
} finally {
lock.unlock();
}
}
private static void sendBreakfast() {
lock.lock();
try {
log.info("送早餐来了");
hasBreakfast = true;
waitbreakfastQueue.signal();
} finally {
lock.unlock();
}
}
}
输出如下：
2023-06-03 17:34:44,999 - 0 INFO [Thread-0] up.cys.chapter03.ReentrantLockTest03:29 - 线程1等待烟
2023-06-03 17:34:45,010 - 11 INFO [Thread-1] up.cys.chapter03.ReentrantLockTest03:47 - 线程2等早餐
2023-06-03 17:34:46,004 - 1005 INFO [main] up.cys.chapter03.ReentrantLockTest03:92 - 送早餐来了
2023-06-03 17:34:46,005 - 1006 INFO [Thread-1] up.cys.chapter03.ReentrantLockTest03:55 - 线程2等到了它的早餐
2023-06-03 17:34:47,011 - 2012 INFO [main] up.cys.chapter03.ReentrantLockTest03:81 - 送烟来了
2023-06-03 17:34:47,013 - 2014 INFO [Thread-0] up.cys.chapter03.ReentrantLockTest03:37 - 线程1等到了它的烟
原理
![]](https://img-blog.csdnimg.cn/4d4dccdf5b084b31a6ad324e5d8b5997.png)
看下上面类图，ReentrantLock实现了Lock接口，同时内部维护了一个同步器Sync。
Sync 继承了 AbstractQueuedSynchronizer ，所以 Sync 就具有了锁的框架，根据 AQS 的框架，Sync 只需要实现 AQS 预留的几个方法即可，但 Sync 也只是实现了部分方法，还有一些交给子类 NonfairSync(非公平锁) 和 FairSync(公平锁) 去实现了。
ReentrantLock内部主要利用CAS+AQS队列来实现。
说明：以下源代码版本为JDK11。
非公平锁实现原理
源码
先从构造器开始看，ReentrantLock默认构造器是非公平锁实现：
public ReentrantLock() {
sync = new NonfairSync();
}
NonfairSync实现的加锁Lock方法实现了加锁，代码如下：
public void lock() {
sync.acquire(1);
}
然后调用了同步器Sync的acquire(1)方法，源代码如下：
public final void acquire(int arg) {
if (!tryAcquire(arg) && // tryAcquire尝试获取锁，如果失败
acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) // 尝试创建一个Node对象，加到等待队列中，如果成功
selfInterrupt(); // 获取失败并且加入队列成功，就调用自己的Interrupt方法
}
然后下面主要看下获取锁tryAcquire方法，tryAcquire是AQS里的抽象方法，找到非公平锁的实现，其源代码如下：
static final class NonfairSync extends Sync {
private static final long serialVersionUID = 7316153563782823691L;
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
}
然后就是进入nonfairTryAcquire方法，源代码如下：
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
// 获取同步器的状态，如果为0，说明锁是空闲的，没有被持有
int c = getState();
if (c == 0) {
// 使用CAS让当前线程持有锁
if (compareAndSetState(0, acquires)) {
// 如果成功，设置exclusiveOwnerThread为当前线程，返回true
setExclusiveOwnerThread(current);
return true;
}
}
// 如果锁已经被占用，比较是否是当前线程占用的，如果是，准备重复加锁，这就是重入锁的原理
else if (current == getExclusiveOwnerThread()) {
// 让计数器加1，表示重入锁上的加锁次数
int nextc = c + acquires;
// int 是有最大值的，如果小于0，说明超过了最大值，直接报错
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
// 设置同步器的状态为nextc
setState(nextc);
return true;
}
// 最后返回加锁失败
return false;
}
回头继续看Sync的acquire(1)方法，当加锁失败，则执行acquireQueued(addWaiter(Node.EXCLUSIVE), arg))方法。
首先是addWaiter(Node.EXCLUSIVE), arg)方法，他返回一个Node对象，内部维护的是一个双向链表
private Node addWaiter(Node mode) {
// 初始化节点
Node node = new Node(mode);
// 死循环，知道把新的node添加到了队列中
for (;;) {
// 获取队列尾节点
Node oldTail = tail;
// 如果为节点不为null，则把新的node节点添加到尾部
if (oldTail != null) {
node.setPrevRelaxed(oldTail);
// 设置新节点为尾节点
if (compareAndSetTail(oldTail, node)) {
oldTail.next = node;
return node;
}
// 如果尾结点是null，则调用初始化队列的方法，并把新节点放入尾部
} else {
initializeSyncQueue();
}
}
}
队列创建完之后作为方法acquireQueued的参数，
final boolean acquireQueued(final Node node, int arg) {
boolean interrupted = false;
try {
// 有一个死循环一直尝试获取锁
for (;;) {
// 获取前驱节点p
final Node p = node.predecessor();
// 如果前驱节点是头节点，说明自己是第二个节点，那么便有资格去尝试获取锁
if (p == head && tryAcquire(arg)) {
setHead(node); // 获取成功,将当前节点设置为head节点
p.next = null; // help GC
return interrupted; //返回interrupted中断过状态
}
// 判断获取失败后是否可以挂起
// 注意第一个参数是前驱节点，方法会前驱节点的状态设为-1，表示他可以用来唤醒后继节点
if (shouldParkAfterFailedAcquire(p, node))
// 若可以,则挂起，并设置interrupted中断状态
interrupted |= parkAndCheckInterrupt();
}
} catch (Throwable t) {
cancelAcquire(node);
if (interrupted)
selfInterrupt();
throw t;
}
}
上面是获取锁的源码。
下面看看释放锁的源码。
首先是unlock方法，调用了sync的release方法：
public void unlock() {
sync.release(1);
}
release方法代码如下：
public final boolean release(int arg) {
// 调用tryRelease方法来释放锁
if (tryRelease(arg)) {
Node h = head;
// 如果释放成功，则检查头部是否不为null，并且头部节点不为0
if (h != null && h.waitStatus != 0)
// 如果是，则唤醒下一个节点
unparkSuccessor(h);
return true;
}
return false;
}
里面主要调用了tryRelease方法：
@ReservedStackAccess
protected final boolean tryRelease(int releases) {
// 计算当前的状态和要释放的数字的差，主要是因为可重入锁可以多次获取锁，释放时也要释放和加锁的次数一样
int c = getState() - releases;
// 如果当前线程不是锁的持有者就报错
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false; // 是否释放成功了，默认为fasle
// 如果c为0，说明锁可以被释放了
if (c == 0) {
free = true; // 释放成功了
// 把锁Owner设为null
setExclusiveOwnerThread(null);
}
// 设置status
setState(c);
return free;
}
流程
根据上面的源码，看下整个加锁解锁流程是什么样的。
流程如下：
- 第一次加锁
第一个线程Thread-0加锁时，没有竞争，直接调用Lock方法成功，会把exclusiveOwnerThread为Thread-0
- 第二个线程Thread-1想要加锁时，如下
- 进入nonfairTryAcquire方法
- getState()获取状态，发现结果是1
- 然后比较当前线程不是持有锁的线程，加锁失败
- 接下来进入 addWaiter 逻辑，构造 Node 队列。图中黄色三角表示该 Node 的 waitStatus 状态，其中 0 为默认正常状态；Node 的创建是懒惰的；其中第一个 Node 称为 Dummy（哑元）或哨兵，用来占位，并不关联线程。
- 当前线程进入 acquireQueued 逻辑
- acquireQueued 会在一个死循环中不断尝试获得锁，失败后进入 park 阻塞
- 如果自己是紧邻着 head（排第二位），那么再次 tryAcquire 尝试获取锁，当然这时 state 仍为 1，失败
- 进入 shouldParkAfterFailedAcquire 逻辑，将前驱 node，即 head 的 waitStatus 改为 -1，这次返回 false
-
shouldParkAfterFailedAcquire 执行完毕回到 acquireQueued ，再次 tryAcquire 尝试获取锁，当然这时state 仍为 1，失败
-
当再次进入 shouldParkAfterFailedAcquire 时，这时因为其前驱 node 的 waitStatus 已经是 -1，这次方法返回true
-
进入 parkAndCheckInterrupt，就挂起当前线程并检查Interrupt状态， 线程Thread-1 状态就为park了（灰色表示）
- 再次有多个线程经历上述过程竞争失败，变成下面这个样子
- 当Thread-0 释放锁，进入 tryRelease 流程，如果成功，设置 exclusiveOwnerThread 为 null，state = 0
- 当前队列不为 null，并且 head 的 waitStatus = -1，进入 unparkSuccessor 流程，找到队列中离 head 最近的一个 Node（没取消的，取消的为-1），unpark 恢复其运行，本例中即为 Thread-1。
- 回到 Thread-1 的 acquireQueued 流程
如果加锁成功（没有竞争），会设置:
- exclusiveOwnerThread 为 Thread-1，state = 1
- head 指向刚刚 Thread-1 所在的 Node，该 Node 清空 Thread
- 原本的 head 因为从链表断开，而可被垃圾回收
-
如果这时候有其它线程来竞争（非公平的体现）
例如这时有 Thread-4 来了
如果不巧又被 Thread-4 占了先:
- Thread-4 被设置为 exclusiveOwnerThread，state = 1
- Thread-1 再次进入 acquireQueued 流程，获取锁失败，重新进入 park 阻塞
锁重入原理
锁的重入原理在上面的源码上就可以看到，主要体现在以下两个方面。
第一个是在上锁的时候，回顾下源码如下：
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
// 获取同步器的状态，如果为0，说明锁是空闲的，没有被持有
int c = getState();
if (c == 0) {
// 使用CAS让当前线程持有锁
if (compareAndSetState(0, acquires)) {
// 如果成功，设置exclusiveOwnerThread为当前线程，返回true
setExclusiveOwnerThread(current);
return true;
}
}
// 如果锁已经被占用，比较是否是当前线程占用的，如果是，准备重复加锁，这就是重入锁的原理
else if (current == getExclusiveOwnerThread()) {
// 让计数器加1，表示重入锁上的加锁次数
int nextc = c + acquires;
// int 是有最大值的，如果小于0，说明超过了最大值，直接报错
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
// 设置同步器的状态为nextc
setState(nextc);
return true;
}
// 最后返回加锁失败
return false;
}
重点在执行else if时，也就是：
上锁时比较了当前线程是不是锁的持有者，如果是，则会把状态加上acquires，从而记录了锁重入的次数。
然后看下释放锁的时候，回顾下源码如下：
@ReservedStackAccess
protected final boolean tryRelease(int releases) {
// 计算当前的状态和要释放的数字的差，主要是因为可重入锁可以多次获取锁，释放时也要释放和加锁的次数一样
int c = getState() - releases;
// 如果当前线程不是锁的持有者就报错
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false; // 是否释放成功了，默认为fasle
// 如果c为0，说明锁可以被释放了
if (c == 0) {
free = true; // 释放成功了
// 把锁Owner设为null
setExclusiveOwnerThread(null);
}
// 设置status
setState(c);
return free;
}
重点看下方法的第一行，和上锁时相反：
在释放锁时，会把状态status减去releases，也就是获取的次数减去要释放的次数，差就是还剩余的重入的次数
可打断原理与不可打断原理
首先上我们上面的源码Lock是默认是不可打断的，回顾acquireQueued方法源码如下：
final boolean acquireQueued(final Node node, int arg) {
boolean interrupted = false;
try {
// 有一个死循环一直尝试获取锁
for (;;) {
// 获取前驱节点p
final Node p = node.predecessor();
// 如果前驱节点是头节点，说明自己是第二个节点，那么便有资格去尝试获取锁
if (p == head && tryAcquire(arg)) {
setHead(node); // 获取成功,将当前节点设置为head节点
p.next = null; // help GC
return interrupted; //返回interrupted中断过状态
}
// 判断获取失败后是否可以挂起
// 注意第一个参数是前驱节点，方法会前驱节点的状态设为-1，表示他可以用来唤醒后继节点
if (shouldParkAfterFailedAcquire(p, node))
// 若可以,则挂起，并设置interrupted中断状态
interrupted |= parkAndCheckInterrupt();
}
} catch (Throwable t) {
cancelAcquire(node);
if (interrupted)
selfInterrupt();
throw t;
}
}
首先如果获取锁失败，县城会被park住，然后并设置interrupted中断状态，但是此时并没有返回interrupted的状态，还会继续进入循环，直到获取到了锁，才把interrupted状态返回了。
执行完acquireQueued方法返回true后，执行了selfInterrupt方法，才产生了中断。代码如下：
public final void acquire(int arg) {
if (!tryAcquire(arg) && // tryAcquire尝试获取锁，如果失败
acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) // 尝试创建一个Node对象，加到等待队列中，如果成功
selfInterrupt(); // 获取失败并且加入队列成功，就调用自己的Interrupt方法,才产生了中断
}
前面我们知道，可打断，需要获取锁时需要使用lock.lockInterruptibly()代替lock.lock()。
lock.lockInterruptibly代码如下：
public void lockInterruptibly() throws InterruptedException {
sync.acquireInterruptibly(1);
}
调用了sync的acquireInterruptibly方法，代码如下：
public final void acquireInterruptibly(int arg)
throws InterruptedException {
// 如果打断标记为真，抛出打断异常
if (Thread.interrupted())
throw new InterruptedException();
if (!tryAcquire(arg))
// 获取锁
doAcquireInterruptibly(arg);
}
其中doAcquireInterruptibly方法代码如下：
private void doAcquireInterruptibly(int arg)
throws InterruptedException {
final Node node = addWaiter(Node.EXCLUSIVE);
try {
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
return;
}
// 上面代码与不可打断模式相同，关键在下面
// 如果判断获取锁失败后可以挂起，并且检查状态是可打断，就直接抛出InterruptedException异常了
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} catch (Throwable t) {
cancelAcquire(node);
throw t;
}
}
可以看到大部分代码与不可打断模式一样，唯一不一样是当线程获取锁失败后可以挂起，并且可打断，就直接抛出异常了。
公平锁原理
首先看下我们上面说的非公平锁的源码：
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
// 获取同步器的状态，如果为0，说明锁是空闲的，没有被持有
int c = getState();
if (c == 0) {
// 使用CAS让当前线程持有锁
if (compareAndSetState(0, acquires)) {
// 如果成功，设置exclusiveOwnerThread为当前线程，返回true
setExclusiveOwnerThread(current);
return true;
}
}
// 如果锁已经被占用，比较是否是当前线程占用的，如果是，准备重复加锁，这就是重入锁的原理
else if (current == getExclusiveOwnerThread()) {
// 让计数器加1，表示重入锁上的加锁次数
int nextc = c + acquires;
// int 是有最大值的，如果小于0，说明超过了最大值，直接报错
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
// 设置同步器的状态为nextc
setState(nextc);
return true;
}
// 最后返回加锁失败
return false;
}
非公平主要体现在当c==0时：
使用CAS让当前线程尝试持有锁，而不会检查AQS队列。
然后看下公平锁的TryAcquire方法源码：
@ReservedStackAccess
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
// 这里是主要代码
// 上锁时，会先检查是否有前驱节点，没有的话使用CAS尝试竞争锁
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;
}
看上面的源码中的关键代码注释：
公平锁上锁时，会使用hasQueuedPredecessors方法先检查是否有前驱节点，没有的话使用CAS尝试竞争锁
hasQueuedPredecessors方法是从AQS继承而来的，看下源码：
public final boolean hasQueuedPredecessors() {
Node h, s;
if ((h = head) != null) { // 如果头不是null
// 再看下第二个节点是否为null
// 如果第二个节点是为null，或者第二个节点的状态为大于0，说明被取消了
if ((s = h.next) == null || s.waitStatus > 0) {
s = null; // traverse in case of concurrent cancellation
// 否则按照FIFO原则寻找最先入队列的并且没有被Cancel的Node ,赋值给s
for (Node p = tail; p != h && p != null; p = p.prev) {
if (p.waitStatus <= 0)
s = p;
}
}
// 如果s节点不是null，并且不是当前的线程，则返回true，说明有前驱节点
if (s != null && s.thread != Thread.currentThread())
return true;
}
// 如果头是null，说明队列为空，返回fasle，说明无前驱节点
return false;
}
条件变量原理
每个条件变量其实就对应着一个等待队列，其实现类是 ConditionObject
public class ConditionObject implements Condition, java.io.Serializable {
private static final long serialVersionUID = 1173984872572414699L;
/** First node of condition queue. */
private transient Node firstWaiter;
/** Last node of condition queue. */
private transient Node lastWaiter;
/**
* Creates a new {@code ConditionObject} instance.
*/
public ConditionObject() { }
// 省略代码
}
其内部也维护了两个变量，firstWaiter和lastWaiter，用来维护在条件变量上等待的对列的头和尾。
await流程
await方法用来把线程加入到条件变量的等待队列。
源码如下：
public final void await() throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
// 添加一个 Node 至等待队列
Node node = addConditionWaiter();
// 释放节点持有的锁，因为可能有重入
int savedState = fullyRelease(node);
int interruptMode = 0;
// 如果该节点还没有转移至 AQS 队列, 就阻塞
while (!isOnSyncQueue(node)) {
// park当前线程，等待被唤醒
LockSupport.park(this);
// 如果被打断, 退出等待队列
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
// 退出等待队列后, 还需要获得 AQS 队列的锁
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null)
// 所有已取消的 Node 从队列链表删除,
unlinkCancelledWaiters();
if (interruptMode != 0)
// 应用打断模式
reportInterruptAfterWait(interruptMode);
}
其中addConditionWaiter源码如下：
private Node addConditionWaiter() {
// 如果不是锁的持有者，直接报错
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node t = lastWaiter;
// 如果最后一个节点是null，就从队列清除
if (t != null && t.waitStatus != Node.CONDITION) {
unlinkCancelledWaiters();
t = lastWaiter;
}
// 创建一个关联当前线程的新 Node,
Node node = new Node(Node.CONDITION);
// 如果为节点是null，说明队列是空的，则把新节点作为头节点，否则把新节点加到t的下个节点
if (t == null)
firstWaiter = node;
else
t.nextWaiter = node;
// 把新节点设为尾节点
lastWaiter = node;
return node;
}
整个流程如下；
-
开始 Thread-0 持有锁，调用 await
进入 ConditionObject 的 addConditionWaiter 流程，创建新的 Node 状态为 -2（Node.CONDITION），关联 Thread-0，加入等待队列尾部
- 接下来进入 AQS 的 fullyRelease 流程，释放同步器上的锁
- unpark AQS 队列中的下一个节点，竞争锁，假设没有其他竞争线程，那么 Thread-1 竞争成功
- park阻塞 Thread-0线程，等待被唤醒
signal流程
signal用来唤醒在当前条件变量等待的线程。
源码如下：
public final void signal() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node first = firstWaiter;
if (first != null)
doSignal(first);
}
其主要方法是执行doSignal，源码如下：
private void doSignal(Node first) {
do {
// 已经是尾节点了
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
first.nextWaiter = null;
} while (!transferForSignal(first) && // 将等待队列中的 Node 转移至 AQS 队列, 如果不成功
(first = firstWaiter) != null); // 且还有节点则继续循环
}
其中主要是transferForSignal方法，用来将等待队列中的 Node 转移至 AQS 队，源码如下：
final boolean transferForSignal(Node node) {
// 如果状态已经不是 Node.CONDITION, 说明节点的线程已经被取消了
if (!node.compareAndSetWaitStatus(Node.CONDITION, 0))
return false;
/*
* Splice onto queue and try to set waitStatus of predecessor to
* indicate that thread is (probably) waiting. If cancelled or
* attempt to set waitStatus fails, wake up to resync (in which
* case the waitStatus can be transiently and harmlessly wrong).
*/
// 加入 AQS 队列尾部，返回的p是原来的尾部，即现在的尾节点的上一个节点
Node p = enq(node);
int ws = p.waitStatus; // 获取p的状态
// 如果ws > 0 ，即上一个节点被取消，或者上一个节点p不能设置状态为 Node.SIGNAL
if (ws > 0 || !p.compareAndSetWaitStatus(ws, Node.SIGNAL))
// 那么就使用unpark 取消当前线程阻塞, 让线程重新同步状态
LockSupport.unpark(node.thread);
return true;
}
整个流程如下：
- 假设 Thread-1 要来唤醒 Thread-0
- 进入 ConditionObject 的 doSignal 流程，取得等待队列中第一个 Node，即 Thread-0 所在 Node
- 执行 transferForSignal 流程，将该 Node 加入 AQS 队列尾部，将 Thread-0 的 waitStatus 改为 0，Thread-3 的waitStatus 改为 -1
改为-1是让他有资格唤醒下个节点
- 最后Thread-1 释放锁，进入 unlock 流程
