并发编程系列-同步器实现一 ReentrantLock
前言
第一篇中介绍了AQS，下面学习记录ReentrantLock的实现及其源码剖析，其中包括AQS核心的源码解读及ReentrantLock的源码解读。
ReentrantLock是一种排它锁的实现，同时仅一个线程可访问，其它线程竞争资源的话会进入同步队列进行等待，包括公平与非公平方式。
同时ReentrantLock支持条件等待及唤醒，会在下一篇介绍。
系列文章
一、并发编程系列-同步器 AQS
二、并发编程系列-同步器实现一 ReentrantLock
三、并发编程系列-同步器实现二 ReentrantLock Condition
四、并发编程系列-同步器实现三 CountDownLatch
五、并发编程系列-同步器实现四 Semaphore
六、并发编程系列-同步器实现五 CyclicBarrier
ReentrantLock
内部实现了一个同步类 Sync extends AbstractQueuedSynchronizer，继承AQS同步器：
ReentrantLock的内部类如下:
abstract static class Sync extends AbstractQueuedSynchronizer {
// 获取锁的抽象方法。为什么抽象往下看
abstract void lock();
// 非公平的获取锁
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
// cas尝试，成功的话更新锁的持有线程
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;
}
// 关键：释放锁，该访问是实现了AQS的tryRelease方法的
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;
}
// 获取条件，对应AQS的条件队列
final ConditionObject newCondition() {
return new ConditionObject();
}
}
Sync是个抽象类，有两个实现：公平锁 FairSync、非公平锁 NonFairSync
static final class NonfairSync extends Sync {
// 非公平获取锁，不用排队，来到就可以试试
final void lock() {
// 尝试修改AQS的state从0到1，成功的话表示获取同步锁成功，并设置当前锁持有线程
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
// 否则调用AQS的竞争锁方法
acquire(1);
}
protected final boolean tryAcquire(int acquires) {
// 调用Sync的非同步锁尝试获取，实现见Sync
return nonfairTryAcquire(acquires);
}
}
static final class FairSync extends Sync {
private static final long serialVersionUID = -3000897897090466540L;
// 公平锁，没有尝试修改状态，直接获取锁
final void lock() {
// 内部会调用下面的tryAcquire(int acquires)方法
acquire(1);
}
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
// 注意：hasQueuedPredecessors() 与非公平锁的唯一区别的地方，对于公平锁，如果队列有节点，直接跳过尝试获取资源。
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;
}
}
// Sync为成员变量，核心的方法都在Sync实现了
private final Sync sync;
// 构造器，默认为非公平锁；有参构造器可以指定
public ReentrantLock() {
sync = new NonfairSync();
}
// 加锁
public void lock() {
sync.lock();
}
// 加锁 可中断，线程被中断会抛异常
public void lockInterruptibly() throws InterruptedException {
sync.acquireInterruptibly(1);
}
// 尝试获取锁，不会进入AQS同步器队列，仅尝试cas state，成功与失败都会立马返回
public boolean tryLock() {
return sync.nonfairTryAcquire(1);
}
// 释放锁
public void unlock() {
sync.release(1);
}
// 获取条件对象
public Condition newCondition() {
return sync.newCondition();
}
//// 其它get set先跳过
使用案例
很简单的一个demo，两个线程(称线程a或线程1、线程b或线程2)；线程a获取锁之后执行任务，线程b进入同步队列等待；
public static void testReentrantLock() throws InterruptedException {
ReentrantLock reentrantLock = new ReentrantLock();
new Thread(() -> {
System.out.println("我是线程1开始竞争锁.");
// 排它锁
reentrantLock.lock();
try {
System.out.println("我是线程1获取锁成功了, 开始执行任务，很久..");
Thread.sleep(20 * 1000);
} catch (InterruptedException e) { e.printStackTrace(); }
// 执行任务结束，释放锁，工作时间变短，可以看释放锁步骤
System.out.println("我是线程1完成了要释放锁...");
// 见 代码块2
reentrantLock.unlock();
}, "a").start();
new Thread(() -> {
try {
Thread.sleep(2000); // 为了让线程a先获取到资源
} catch (InterruptedException e) { e.printStackTrace(); }
System.out.println("我是线程2开始竞争锁.");
// 见 代码块1 排它锁，跟进查看获取锁进队列逻辑在此
reentrantLock.lock();
try {
System.out.println("我是线程2获取锁成功了, 开始执行任务..");
Thread.sleep(5000);
} catch (InterruptedException e) { e.printStackTrace(); }
System.out.println("我是线程2完成了要释放锁...");
reentrantLock.unlock();
}, "b").start();
// 主线程睡眠等待
Thread.sleep(5 * 60 * 60 * 1000);
}
上来先让线程b睡眠2s，保证线程a先获取到锁。
输出结果：
我是线程1开始竞争锁.
我是线程1获取锁成功了, 开始执行任务，很久..
我是线程2开始竞争锁.
我是线程1完成了要释放锁...
我是线程2获取锁成功了, 开始执行任务..
我是线程2完成了要释放锁...
案例源码分析
先通过流程图大致看下过程：
具体步骤：
代码块1
在线程1已经获取锁资源的情况下，线程2竞争锁资源
// 获取锁
java.util.concurrent.locks.ReentrantLock.NonfairSync
final void lock() {
// cas尝试获取资源，案例中线程1会获取成功，直接返回；线程2会进入esle逻辑，竞争锁资源。这里主要看线程2的逻辑
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
// 进入AQS竞争锁，继续看
acquire(1);
}
java.util.concurrent.locks.AbstractQueuedSynchronizer
public final void acquire(int arg) {
// 此处的tryAcquire(arg)会尝试cas资源，由AQS子类实现ReentrantLock实现的，继续看
// 1、如果竞争成功则直接返回
// 2、否则调用addWaiter(Node.EXCLUSIVE)，创建一个排它锁节点
// 3、再调用acquireQueued进入队列
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
继续看 addWaiter(Node.EXCLUSIVE) 创建排它锁节点
private Node addWaiter(Node mode) {
// 创建节点，传入当前线程，表明线程与节点的对应，mode是排它锁的标识
Node node = new Node(Thread.currentThread(), mode);
Node pred = tail;
// 如果尾节点不空
if (pred != null) {
// 将当前线程2所在节点的前继节点指向尾节点
node.prev = pred;
// cas将当前线程2所在节点设置成尾节点，成功的话则返回
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
// 上面cas操作失败的情况下，会通过轮训不断尝试直至成功，并返回节点，当前案例会进入当前方法，因为线程1虽然持有锁，但是没有队列
enq(node);
return node;
}
// 不断轮训设置尾节点的操作
private Node enq(final Node node) {
for (;;) { // 无限循环
Node t = tail;
if (t == null) { // 没有尾节点，cas一个新节点作为头节点，并且将尾节点也指向它，注意：该节点没有对应的线程，可以看作是线程1的
if (compareAndSetHead(new Node()))
tail = head;
} else {
// 再一次循环到此，将线程2的节点设置成尾节点，直至成功。
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
继续看，现在已经通过 addWaiter(Node.EXCLUSIVE) 创建排它锁节点，继续调用acquireQueued进入队列：
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
// 中断标识，但如果中断也不会抛异常
boolean interrupted = false;
for (;;) { // 循环
// 获取线程2节点的前继节点
final Node p = node.predecessor();
// 如果p节点是头节点，此例是的，所以会再次尝试获取下锁，哈哈，真是不放过一次机会去尝试，如果刚巧线程1此刻执行完任务释放了锁，真可以成功
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
// 线程1还没有执行完任务，所以会进入到这里
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
// parkAndCheckInterrupt()调用LockSupport.park(this);进入线程等待状态，等待唤醒，并返回是否是中断的（可能唤醒的情况 1、unpark(线程)；2、线程interrupt）
interrupted = true;
}
} finally {
if (failed)
// 如果中断/异常，会进行取消节点
cancelAcquire(node);
}
}
继续看 shouldParkAfterFailedAcquire(p, node)
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
// 获取前继节点的等待状态，AQS那章已经说过，默认是0，会经过下面的cas改成SIGNAL(-1)
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
// 如果是-1，表示后继节点可以安心睡觉了，当前节点锁释放后会唤醒它的
return true;
if (ws > 0) {
// ws大于1表示节点状态已经取消了，可以跳过该节点了
do {
// 比较难看懂，从后往前看，pred = pred.prev表示前节点指到再前一个节点；node.prev = pred当前node节点的前节点指向刚刚的pred。加上后面那句pred.next = node; 其实就是删除中间的取消节点。
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
// 将waitStatue cas成signal状态
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
继续看，shouldParkAfterFailedAcquire经过2次调用，最终会返回true，紧接着调用parkAndCheckInterrupt()，进入睡眠状态
private final boolean parkAndCheckInterrupt() {
// 进入awit，正常唤醒LockSupport.unpark(线程)
LockSupport.park(this);
return Thread.interrupted(); // 是否是中断返回
}
至此，线程2可以安心睡大觉，等待队列上一个节点(线程1)的唤醒。
下面看线程1执行完任务，开始唤醒现场2了
代码块2
java.util.concurrent.locks.ReentrantLock
public void unlock() {
sync.release(1); // 核心方法调用
}
继续看，进入AQS类操作
java.util.concurrent.locks.AbstractQueuedSynchronizer
public final boolean release(int arg) {
if (tryRelease(arg)) { // 尝试释放锁，由AQS子类ReentrantLock实现
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h); // 唤醒h节点线程
return true;
}
return false;
}
继续看，释放不分公平与非公平，都是Sync的方法
java.util.concurrent.locks.ReentrantLock#Sync
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);
}
// 不用cas更新状态，会成功，因为当前线程是持有排它锁的
setState(c);
return free;
}
继续看，tryRelease()成功的话，会获取头节点，如果队列有节点，会继续调用unparkSuccessor(h)，我们知道还是有节点线程2等待的。
java.util.concurrent.locks.AbstractQueuedSynchronizer
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;
if (ws < 0)
// 更新为0 不太清楚为了啥，可能是只操作一次，进行复位吧！
compareAndSetWaitStatus(node, ws, 0);
// 注意：获取头节点的下一个节点进行唤醒的，因为头节点是持有锁的节点。
Node s = node.next;
// s.waitStatus表示已经被取消了，会循环从后到前，找到第一个等待中的线程节点
if (s == null || s.waitStatus > 0) {
s = null;
for (Node t = tail; t != null && t != node; t = t.prev)
if (t.waitStatus <= 0)
s = t;
}
if (s != null)
// 进行唤醒（此处，是唤醒了线程2的）
LockSupport.unpark(s.thread);
}
至此，线程1释放锁并唤醒线程2成功。
最后还差一哆嗦，唤醒了等待的节点线程2，线程2从等待的地方被唤醒，如下：
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
// 中断标识，但不会抛异常
boolean interrupted = false;
for (;;) { // 循环，被唤醒后会继续进入循环，就可以通过调用tryAcquire(arg)成功了，将当前节点更新为头节点，并释放前节点帮助GC；另外注意如果是中断的也会继续获取锁资源的
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) &&
// 上次等待的地方
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
开心，唤醒的线程，获取锁成功，可以执行任务了。
至此，通过源码分析了线程2如何竞争锁失败进入队列的；线程1释放锁的时候，如何唤醒线程2的。
总结几个点
1、第一个获取排它锁的线程是不会创建队列的，仅有一个锁持有标识，获取getExclusiveOwnerThread();
2、第一个线程获取锁后，是没有队列的；再有线程竞争时，锁已经被其它线程持有了，所以此时会创建两个节点，一个头节点，没有线程数据，一个是当前线程的节点，追加到头节点之后，有线程数据。可以把头节点当作持有锁的那个线程节点，因为头节点就是表示持有锁的。
本文是原创文章，采用 CC BY-NC-ND 4.0 协议，完整转载请注明来自 蚂蚁工匠
阅读建议
评论
匿名评论
隐私政策
你无需删除空行，直接评论以获取最佳展示效果