基于ReentrantLock的实现原理讲解
目录
- ReentrantLock实现核心–AQS(AbstractQueuedSynchronizer)
- Node结构
- ReentrantLock实现分析
- 二者关联
- NonfairSync分析
- FairSync分析
- 注意一下
java.util.concurrent包中的工具实现核心都是AQS，了解ReentrantLock的实现原理，需要先分析AQS以及AQS与ReentrantLock的关系。
这篇文章中分析了ReentrantLock#lock与ReentrantLock#unlock的实现，对于Condition的实现分析，另外文章再讲，基本上大同小异。
ReentrantLock实现核心–AQS(AbstractQueuedSynchronizer)
AQS，队列同步器，在juc包中的工具类都是依赖于AQS来实现同步控制，看一张AQS的结构图。
同步控制中主要使用到的信息如上图所示。AQS可以被当做是一个同步监视器的实现，并且具有排队功能。当线程尝试获取AQS的锁时，如果AQS已经被别的线程获取锁，那么将会新建一个Node节点，并且加入到AQS的等待队列中，这个队列也由AQS本身自己维护。当锁被释放时，唤醒下一个节点尝试获取锁。
- 变量exclusiveOwnerThread在互斥模式下，表示当前持有锁的线程。
- 变量tail指向等待获取AQS的锁的节点队列的最后一个
- 变量head指向队列中head节点，head节点信息为空，不表示任何正在等待的线程。
- 变量state表示AQS同步器的状态，在不同情况下含义可能不太一样，例如以下几种
- 在ReentrantLock中，表示AQS的锁是否已经被占用获取，0：没有，>=1：已被获取,当大于1时表示被同一线程多次重入锁。
- 在CountDownLatch中，表示计数还剩的次数，当到达0时，唤醒等待线程。
- 在Semaphore中，表示AQS还可以被获取锁的次数，获取一次就减1，当到达0时，尝试获取的线程将会阻塞。
Node结构
Node节点是AQS管理的等待队列的节点元素，除了head节点之外，其他一个节点就代表一个正在等待线程的队列。Node一般的重要参数有几个。
- prev 前置节点
- next后置节点
- thread 代表的线程
- waitStatus节点的等待状态
- 1表示节点已经取消，也就是线程可能已经中断，不需要再等待获取锁了，在后续代码中会处理跳过waitStatus等于1的节点
- -1表示当前节点的后置节点代表的线程需要被挂起
- -2表示当前线程正在等待的是Condition锁
ReentrantLock实现分析
二者关联
ReentrantLock实现核心是基于AQS,看下面一张图，分析AQS与ReentrantLock的关系。
从图中可以看出，ReentrantLock里面有最终两个内部类，FairSync和NonfairSync通过继承AbstractQueuedSynchronizer的功能，来实现两种同步互斥方案：公平锁和非公平锁。
在ReentrantLock中最终lock和unlock操作，都由FairSync和NonfairSync实际完成。
NonfairSync分析
下面看一个最简单利用ReentrantLock实现互斥的例子。
ReentrantLock lock = new ReentrantLock();
//尝试获取锁
lock.lock();
//获得锁后执行逻辑......
//......
//解锁
lock.unlock();
在ReentrantLock的默认无参构造方法中，创建的是非公平锁
public ReentrantLock() {
sync = new NonfairSync();
}
下面分析lock.lock();这句代码是如何实现同步互斥的。
NonfairSync#lock
点开lock.lock();源码，可以看到最终实际调用的是NonfairSync#lock，这是分析的入口。
NonfairSync#lock源码如下。
final void lock() {
if (compareAndSetState(0, 1))//【step1】
setExclusiveOwnerThread(Thread.currentThread());//【step2】
else
acquire(1);//【step3】
}
【step1】上面有提到，NonfairSync继承自AbstractQueuedSynchronizer，NonfairSync就是一个AQS，因此在步骤【1】其实就是利用CAS（一个原子性的比较并设置操作）尝试设置AQS的state为1。如果设置成功，表示获取锁成功；如果设置失败，表示state之前已经是>=1，已经被别的线程占用了AQS的锁，所示无法设置state为1，稍后会把线程加入到等待队列。
**非公平锁与公平锁：**对于NonfairSync非公平锁来说，线程只要执行lock请求，就会立马尝试获取锁，不会管AQS当前管理的等待队列中有没有正在等待的线程，这种操作是不公平的，没有先来后到；而稍后介绍的FairSync公平锁，则会在lock请求进行时，先判断AQS管理的等待队列中是否已经有正在等待的线程，有的话就是不尝试获取锁，直接进入等待队列，保证了公平性。
这一步需要熟悉的是CAS操作，分析一下compareAndSetState源码，如下。这一步利用unsafe包的cas操作，unsafe包类似一种java留给开发者的后门，可以用来直接操作内存数据，并且保证这个操作的原子性。在下面的代码中，stateOffset表示state比变量的内存偏移地址，用来寻找state变量内存位置。整个cas操作就是找到内存中当前的state变量值，并且与expect期待值比较，如果跟期待值相同，那么表示这个值是可以修改的，此时就会对state值进行更新；如果与期待值不一样，那么将不能进行更新操作。unsafe保证了比较与设置值的过程是原子性的，在这一步不会出现线程安全问题。
protected final boolean compareAndSetState(int expect, int update) {
// See below for intrinsics setup to support this
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}
【step2】操作是在设置state成功之后，表示当前线程获取AQS锁成功，需要设置AQS中的变量exclusiveOwnerThread为当前持有锁的线程，做保存记录。
【step3】当尝试获取锁失败的时候，就需要进行步骤3，执行acquire,进行再次尝试或者线程进入等待队列。
AbstractQueuedSynchronizer#acquire
当第一次尝试获取锁失败之后，执行【step3】acquire方法，这个方法可以讲一个尝试获取锁失败的线程放入AQS管理的等待队列进行等待，并且将线程挂起。实现逻辑在AbstractQueuedSynchronizer实现，NonfairSync通过继承AbstractQueuedSynchronizer获得等待队列管理的功能。
下面看源码。
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
NonfairSync#tryAcquire–锁重入实现
首先，执行tryAcquire再次尝试一次获取lock，tryAcquire是由子类实现，也就是这里调用NonfairSync#tryAcquire方法，跟踪调用，最终执行代码如下。
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
//如果此时state已经变为1，再次尝试一次获取锁
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) {
//否则判断当前持有AQS的锁的线程是不是当前请求获取锁的线程，是的话就进行锁重入。
int nextc = c + acquires;
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
对于NonfairSync#tryAcquire的实现，首先是重新尝试一次获取锁，如果还是获取不到，就尝试判断当前的是不是属于重入锁的情形。
对于重入锁的情形，就需要对state进行累加1，意思就是重入一次就对state加1。这样子，在解锁的时候，每次unlock就对state减一，等到state的值为0的时候，才能唤醒下一个等待线程。
如果获取成功，返回true；否则返回false，继续执行下一步acquireQueued(addWaiter(Node.EXCLUSIVE), arg))，添加一个Node节点进入AQS管理的等待队列。
AbstractQueuedSynchronizer#addWaiter–添加Node到等待队列
这个方法属于队列管理，也是由AbstractQueuedSynchronizer默认实现，NonfairSync继承获得。
查看addWaiter源码实现。
private Node addWaiter(Node mode) {
//新建一个Node节点，mode传入表示当前线程之间竞争是互斥模式
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
//尝试添加当前新建Node到链表队列末尾
Node pred = tail;
if (pred != null) {
node.prev = pred;
//利用cas设置尾指针指向的节点为当前线新建节点
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
//当前是空的没有任何正在等待的线程Node的时候，执行enq(node)，初始化等待队列
enq(node);
return node;
}
private Node enq(final Node node) {
for (;;) {
Node t = tail;
if (t == null) { // Must initialize
//新建一个空的头节点
if (compareAndSetHead(new Node()))
tail = head;
} else {
//跟之前一样，利用cas这只当前新建节点为尾节点
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
以上操作，完成将一个节点加入队列操作。加入完成之后，返回这个新加入的节点Node给acquireQueued方法继续执行。
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))–锁竞争优化
上面既然都完成了等待节点如队列的操作，为什么不直接挂起线程进入等待呢？因此这里的设计者做了一个优化操作。acquireQueued方法其实就是为了减少线程挂起、唤醒次数而作的优化操作。
下面看看acquireQueued的代码实现。
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
//获取当前竞争锁的线程Node的前置节点
final Node p = node.predecessor();
//【step1】
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
//【step2】
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
【step1】假如前置节点是head，那么表示当前线程是等待队列中最大优先级的等待线程，可以继续最后的尝试获取锁，因为很有可能会获取到锁，从而避免线程挂起、唤醒，耗费资源，这里也算是最终一次尝试获取。
【step2】shouldParkAfterFailedAcquire(p, node)是检查当前是否有必要挂起，前面我们说过，只有当前置节点的waitStatus是-1的时候才会挂起当前节点代表的线程。当shouldParkAfterFailedAcquire(p, node)返回true的时候，就可以执行parkAndCheckInterrupt()来挂起线程。
shouldParkAfterFailedAcquire(p, node)源码如下。
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
//前置节点是-1，返回true表示线程可挂起
return true;
if (ws > 0) {
//前置节点大于0表示前置节点已经取消，那么进行跳过前置节点的操作，做链表的基本删除节点操作
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
//如果前置节点还是0,表示前置节点Node的waitStatus是初始值，需要设置为-1，然后外层循环重新执行shouldParkAfterFailedAcquire方法，即可挂起当前线程。
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
private final boolean parkAndCheckInterrupt() {
//阻塞挂起线程，等待唤醒
LockSupport.park(this);
//唤醒后，重置中断标记，线程中断标记位为不中断
return Thread.interrupted();
}
FairSync分析
之前提到
**非公平锁与公平锁：**对于NonfairSync非公平锁来说，线程只要执行lock请求，就会立马尝试获取锁，不会管AQS当前管理的等待队列中有没有正在等待的线程，这种操作是不公平的，没有先来后到；而稍后介绍的FairSync公平锁，则会在lock请求进行时，先判断AQS管理的等待队列中是否已经有正在等待的线程，有的话就是不尝试获取锁，直接进入等待队列，保证了公平性。
所以两者的实现区别在于第一次尝试lock的动作不一样。
FairSync#lock
final void lock() {
acquire(1);
}
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
最终差异体现在FairSync#tryAcquire的实现（第一次尝试获取lock）
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
//hasQueuedPredecessors判断队列是否还有别的线程在等待锁，没有的话就尝试获取lock
//如果有别的线程在等待锁，就不会尝试获取lock；下面如果也不是重入的情况的话就直接进入等待队列
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;
}
AbstractQueuedSynchronizer#release --AQS解锁操作
AQS中定义了解锁操作的模板方法，tryRelease(arg)是不同AQS子类实现,对state的多样化操作。例如ReentrantLock中的tryRelease(arg)操作比较明显的就是对state减一。
tryRelease(arg)返回结果表示本次操作后是否需要唤醒下一个等待节点。对于ReentrantLock就是state减一之后是否变为0了。如果需要唤醒下一个节点的线程，那么判断一下Head有没有下一个要唤醒的节点线程，有的话就进行唤醒操作unparkSuccessor(h);
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
ReentrantLock.Sync#tryRelease–解锁实现
看一下ReentrantLock.Sync的tryRelease实现.是如何为state减一 的。。
protected final boolean tryRelease(int releases) {
//获取state减掉realease，对于ReentrantLock就是默认减一
int c = getState() - releases;
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
if (c == 0) {
//如果减到0了，那么久释放锁
free = true;
//设置持有线程为null
setExclusiveOwnerThread(null);
}
//设置state为新的
setState(c);
return free;
}
至于这里设置state为啥不同cas操作，因为
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
所以永远只有持有锁的线程会做解锁减一的操作，state设置是线程安全的。
注意一下
其实这里还没分析Condition的实现原理，篇幅太长，下次另开文章分析。以上为个人经验，希望能给大家一个参考，也希望大家多多支持自由互联。
【来源：海外服务器https://www.68idc.cn】