您的位置：首页 > 编程语言

并发编程之深入理解ReentrantLock和AQS原理

2020-07-18 20:27 423 查看

AQS（AbstractQueuedSynchronizer）在并发编程中占有很重要的地位，可能很多人在平时的开发中并没有看到过它的身影，但是当我们有看过concurrent包一些JDK并发编程的源码的时候，就会发现很多地方都使用了AQS，今天我们一起来学习一下AQS的原理，本文会用通俗易懂的语言描述AQS的原理。当然如果你了解CAS操作、队列、那么我相信你学习起来会感到无比轻松。

我们会从锁（ReentrantLock）的入口来学习AQS，当然AQS不仅仅只是实现了锁，在很多的工具类中（如CountDownLatch、Semaphore），感兴趣的可以去看看，当我们理解了AQS的原理，我们再过去看那些源码真的可以说是so easy！我们开始吧

一、简单思考锁的实现原理

我们今天不讨论synchronized的实现，因为它是从jvm语言层面实现的锁，我们也很难看到它的源码，我们今天重点从JDK的ReentrantLock的实现着手。

锁的实现原理，无非就是限制多个线程执行一段代码块时，每次允许一个线程执行一段代码块，那如果是你来实现锁，你将会如何实现？

我这里假设一下实现的步骤

1、定义一个int类型的state变量（volatile），当state=0（锁没有被线程持有），当state=1（锁被其他线程持有）

2、当线程去抢锁的时候，就是将state=0变成state=1，如果成功则抢到锁

3、当线程释放锁的时候，就是将state=1变成state=0

4、当我们没有没有抢到锁，就进行等待，加入一个队列进行排队

5、加入到队列的线程一直监听锁的状况，当有机会抢到锁的时候，就尝试去抢锁

如果你跟我想的一样，那么恭喜你，实现锁的主要的流程你基本上已经掌握了，JDK主要的思路也是这样子，但是他们的思路比上面更加严谨，具体严谨在哪里呢？我们接着往下看

1、当state=0变成state=1的过程的原子性（因为这个操作类似i++，不是原子性的）

2、锁的可重入性，比如递归调用

3、当没有抢到锁时加入到队列的时候，也要保证原子性，意思就是如果threadA，threadB，threadC同时竞争锁，只有threadA竞争到了，那么要保证threadB和threadC能够同时加入到队列的尾部，不能出错

4、如果处于队列中等待的线程一直与循环监听锁，会不会导致性能下降？还是说当锁释放了，会进行通知唤醒队列中的一个线程。

其实ReentrantLock的锁基本上就很好的解决了上述的问题。

二、JDK中ReentrantLock的实现原理

1、ReentrantLock分公平锁和非公平锁

ReentrantLock通过构造函数中传入boolean类型，用于创建公平锁和非公平锁（默认是非公平锁，因为非公平锁性能相对要高一点）

public ReentrantLock(boolean fair) {
sync = fair ? new FairSync() : new NonfairSync();
}

[/code]

为什么性能会高一点呢？因为非公平锁在调用lock的时候，首先就会去抢一次，如果抢到了就操作。有可能在线程上下文切换的过程中，一个很短的任务抢到锁了刚好在该上下文切换的时间内执行完了任务。如果是公平锁，就会加入到队列的尾部，等待它前面的线程都执行完了，再执行

2、ReentrantLock内部结构

ReentrantLock内部的结构非常简单，这是因为复杂的逻辑封装在了AbstractQueuedSynchronizer中（我们今天的重点，也是难点），下面类图是ReentrantLock内部类的关系图

这里使用了模板设计模式，不了解的可以参考这篇文章（模板设计模式）

我们开始先看下ReentrantLock内部实现上锁和释放锁的逻辑，看看和我们前面自己思考的实现锁的逻辑是不是一致，这里我们以非公平锁为例，我相信非公平锁理解了，公平锁也是so easy的

3、ReentrantLock源码

（1）NonfairSync.lock()

/**
* Performs lock.  Try immediate barge, backing up to normal
* acquire on failure.
*/
final void lock() {
//cas的操作保证原子性
if (compareAndSetState(0, 1))
//设置当前抢到锁的线程
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}

[/code]

lock的代码非常简单，首先尝试使用cas（compare and swap）尝试获取锁，注意这里使用cas没有用到自旋（无限循环，这里只尝试了一次）。跟我们之前想的一样，无非就是将state的值使用cas从0–>1，如果成功，则表示抢到了锁，并且设置当前抢到锁的线程（后面可重入或者释放锁的时候，都需要判断该线程），如果没有抢到就走else的逻辑

（2）AbstractQueuedSunchronizer.acquire()

public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}

[/code]

这里短路与，会再次尝试一次获取，如果没有获取到则加入队列（将当前线程信息封装成Node节点，使用cas加入到队列尾部），我们先看tryAcquire()，加入队列的逻辑到下面一节再说。

这里使用了模板方法，其实调用到了ReentrantLock内部的NonfairSync的tryAcquire()

（3）tryAcquire()

protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}

[/code]

（4）nonfairTryAcquire()

/**
* Performs non-fair tryLock.  tryAcquire is implemented in
* subclasses, but both need nonfair try for trylock method.
*/
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
//获取state值
int c = getState();
//state=0表示当前没有线程持有锁，则使用cas尝试获取
if (c == 0) {
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
//抢到了就退出
return true;
}
}
//如果state>0则表示当前锁被线程持有，则判断是不是自己持有
else if (current == getExclusiveOwnerThread()) {
//如果是当前线程，则重入，state+1
int nextc = c + acquires;
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}

[/code]

这一段逻辑也非常的简单，逻辑如下

1、获取当前的state的值

2、如果state=0，表示当前没线程持有锁，则尝试获取锁（将state=0使用cas修改成1，如果成功则设置当前线程，和上面的逻辑一致）

3、如果state>0表示当前锁被线程持有，则判断持有锁的线程是不是当前线程，如果是当前线程，则state+1，这里是实现可重入锁的关键

4、否则返回false，则会将当前线程的信息生成Node节点，打入到等待队列，后面会讲

相信大家到这里明白了ReentrantLock中lock的第一步了，其实和我们之前想的自己实现锁的方式是一致的，下面我们开始看释放锁的逻辑

（1）unLock()

public void unlock() {
sync.release(1);
}

[/code]

（2）release()

public final boolean release(int arg) {
//释放锁
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}

[/code]

这里我们先简单的看tryRelease()，关于队列的操作，后面会详细单独讲，其实这里就是释放锁，然后队列的下一个节点由阻塞状态变成非阻塞，从名字也能看出来。

这里也是模板方法，进入了ReentrantLock的tryRelease

（3）tryRelease()

protected final boolean tryRelease(int releases) {
//直接将state-releases(允许一次释放多次，比如await方法，就会直接从state=n变成0)
int c = getState() - releases;
//当前释放锁的线程不是持有锁的线程则抛异常
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
if (c == 0) {
free = true;
//将持有锁的线程记录变量置为null
setExclusiveOwnerThread(null);
}
setState(c);
return free;
}

[/code]

这里的逻辑如下：

1、首先将state-releases，这里如果是实现锁的情况，releases的值一般是1，这里我详细解释一下，假如线程A第一次获取锁则state=1，当线程A继续获取该锁（重入）则state+1=2，以此类推，每重入一次则加1，当释放锁的时候，则进行相应的减1，只有当全部释放完state=0时才返回true，但是如果当调用condition.await()方法则会直接将state减成0，因为要全部释放锁

2、判断当前释放锁的线程是不是持有锁的线程，如果不是则抛异常，线程A不能释放线程B持有的锁

3、当全部释放完，state=0，则将持有锁的线程变量设置成null，表示当前没有线程持有锁

4、否则返回false

到这里ReentrantLock上锁和释放锁的逻辑基本上就结束了（还没进入后面主题AQS）

总结：

1、ReentrantLock是通过一个int类型的state去控制锁

2、当state=0表示当前锁没有被占有，>0表示被线程占有

3、抢锁的过程其实就是使用cas尝试讲state=0修改成state=1，如果抢到锁，需要记录抢到锁的线程

4、当一个线程多次获取一个锁时，是在state做累加，同时释放的话就递减、

5、释放锁就是将state=1（或者>1是递减）变成state=0，此时不需要使用cas，因为没有竞争，锁是被当前线程持有的，当锁完全释放，则设置当前持有锁的那个变量设置为null

三、AQS原理

到这里才算真正进入本片文章的主题，前面讲到ReentrantLock是希望大家平滑过渡到AQS，不然直接进来说AQS会比较干，不丝滑。前面我们简单说过当使用cas尝试获取锁时，如果失败会使用cas将当前线程的信息封装成Node节点加入到一个队列的末尾，我们就从这里作为入口，深入AQS

我们先来看下数据结构、并且描述一下同步队列的样子、以及工作的流程、最后再来看代码

1、Node结构

前面简单的说过我们竞争锁的线程信息会被封装到Node中，这里对Node详细解析一下

thread：当前竞争锁的线程

prev：前一个node节点，因为同步队列是一个双向队列

next：后一个node节点，因为同步队列是一个双向队列

waitStatus：当前线程的等待状态，它的值一般就是里面定义的CANCELLED（已经取消）、SIGNAL（准备就绪等待通知唤醒即可）、PROPAGATE（共享锁SHARED用到）、CONDITION（在某个条件上等待）

nextWaiter:是condition的等待队列中用到，下一个等待节点，因为condition使用的等待队列的Node数据结构和AQS同步队列的Node数据结构是同一个

2、队列的结构

一般来说head指向的节点是获取了锁的节点，当它释放锁后，会通知后一个节点（后面的节点可能是处理阻塞的状态，则可能会被唤醒）

大家可以先结合图来看下他们的流程，后面再去看源码可能会轻松很多，抢锁的逻辑大致如下：

（1）当抢锁失败的时候，会将当前的线程信息封装成Node节点使用CAS加入到队列的尾部（因为可能有多个线程同时加入尾部）

（2）加入到队列之后，当前线程会获取前一个节点的信息，如果前一个节点是head节点，则会尝试获取锁，获取到了就会将自己设置成head节点，并且将之前队列的head节点设置成null，让垃圾回收器回收，从当前队列移除；如果前一个节点不是head节点或者获取锁失败则会判断是否进行阻塞，一般会进行阻塞（防止自旋耗费性能）

（3）当head释放锁的时候，会唤醒head的后一个阻塞的节点，此时被唤醒后的节点进入自旋尝试获取锁（因为这个时候并不能保证一定会获取锁，比如前面讲的刚创建的线程会先尝试能不能获取锁，就会产生竞争，这也是为什么非公平锁比公平锁性能好的原因），如果没有获取到则又会进入阻塞等待唤醒

3、深入源码分析

相信结合上面的图，以及上述逻辑的描述，大家已经对整体的逻辑有一定的把握，再来看看源码

先从获取锁失败加入到队列的尾部的源码开始

（1）acquire()

public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}

[/code]

我们在上一节只分析了tryAcqure(arg)没有分析后面，今天我们从这里开始分析。我们先看addWaiter(Node.EXCLUSIVE)后面再看acquireQueued()

（2）addWaiter()

/**
* Creates and enqueues node for current thread and given mode.
*
* @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared
* @return the new node
*/
private Node addWaiter(Node mode) {
//封装线程信息，并且mode为独占锁，ReentrantLock本来就是独占锁
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
Node pred = tail;
if (pred != null) {
node.prev = pred;
//cas设置队尾
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
//自旋cas加入队列
enq(node);
return node;
}

[/code]

这段代码也比较简单，将当前线程信息封装成Node，这里的mode是共享模式还是独占模式（SHARED、EXCLUSIVE），在Node里面能看得到，我们这里先看独占模式EXCLUSIVE。这里首先会尝试使用cas加入到队列的尾部，如果成功则return退出，否则调用enq(node)

（2）enq(node)

/**
* Inserts node into queue, initializing if necessary. See picture above.
* @param node the node to insert
* @return node's predecessor
*/
private Node enq(final Node node) {
for (;;) {
Node t = tail;
if (t == null) { // Must initialize
if (compareAndSetHead(new Node()))
tail = head;
} else {
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}

[/code]

这里就用了自旋（死循环，直至成功）cas将node加入到队列的尾部，当前前面有一个初始化的判断，如果队列没有初始化，则会初始化，到这里没有抢到锁的Node已经成功加入到同步队列的尾部了，后面就是如何让他知道什么时候应该可以去抢锁了。我们接着看acquireQueued(addWaiter(Node.EXCLUSIVE), arg))，上面已经分析了addWaiter方法，现在分析acquireQueued()

（3）acquireQueued()

final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
//自旋
for (;;) {
final Node p = node.predecessor();
//判断当前节点的前一个节点是不是头节点，如果是，则尝试获取一次锁
if (p == head && tryAcquire(arg)) {
//获取成功则将自己设置成头节点
setHead(node);
//将之前的头节点从队列中一处
p.next = null; // help GC
failed = false;
return interrupted;
}
//如果上面获取失败的话，这里就会判断是否需要阻塞，
//主要是防止cpu无限调度这一块自旋代码，降低性能，从而使用通知的模式
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}

[/code]

这里的代码看着也不难理解，很多时候我们可以从方法名就能看到方法的主要意图，上面的注释基本上描述了主要的逻辑，这里就不在继续描述了，我们看一下里面的阻塞的逻辑shouldParkAfterFailedAcquire()

（4）shouldParkAfterFailedAcquire(prev,node)

/**
* Checks and updates status for a node that failed to acquire.
* Returns true if thread should block. This is the main signal
* control in all acquire loops.  Requires that pred == node.prev.
*
* @param pred node's predecessor holding status
* @param node the node
* @return {@code true} if thread should block
*/
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
/*
* This node has already set status asking a release
* to signal it, so it can safely park.
*/
return true;
if (ws > 0) {
/*
* Predecessor was cancelled. Skip over predecessors and
* indicate retry.
*/
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
/*
* waitStatus must be 0 or PROPAGATE.  Indicate that we
* need a signal, but don't park yet.  Caller will need to
* retry to make sure it cannot acquire before parking.
*/
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}

[/code]

这段逻辑主要的目的就是去除node节点的所有的状态为CANCELED的节点，CANCELED表示取消，不再获取锁，否则就阻塞（这里要注意了，当返回false时下次for循环进入到这里时依然会阻塞），阻塞之后就不会调用自旋的for循环耗费cpu了，而是等待前面的Node节点释放锁之后通知唤醒它。到这里获取锁失败，并且加入队列阻塞等待已经分析完了，后面我们分析当前面的Node释放锁时，通知阻塞的Node节点吧。我们直接从release()方法开始吧，release方法是由unlock里面调用的

（5）release()

public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
//唤醒阻塞的后继者
unparkSuccessor(h);
return true;
}
return false;
}

[/code]

tryRelease()前面已经分析过了，这里不继续分析，如果tryRelease已经完成成功释放锁了（state=0）返回true，

则会唤醒阻塞的后一个节点

（6）unparkSuccessor()

/**
* Wakes up node's successor, if one exists.
*
* @param node the node
*/
private void unparkSuccessor(Node node) {
/*
* If status is negative (i.e., possibly needing signal) try
* to clear in anticipation of signalling.  It is OK if this
* fails or if status is changed by waiting thread.
*/
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);

<span class="token comment">/*
* Thread to unpark is held in successor, which is normally
* just the next node.  But if cancelled or apparently null,
* traverse backwards from tail to find the actual
* non-cancelled successor.
*/</span>
Node s <span class="token operator">=</span> node<span class="token punctuation">.</span>next<span class="token punctuation">;</span>
<span class="token comment">//如果存在后继节点，或者后继节点的状态为CANCELLED</span>
<span class="token keyword">if</span> <span class="token punctuation">(</span>s <span class="token operator">==</span> null <span class="token operator">||</span> s<span class="token punctuation">.</span>waitStatus <span class="token operator">&gt;</span> <span class="token number">0</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
s <span class="token operator">=</span> null<span class="token punctuation">;</span>
<span class="token comment">//从尾部开始取需要被唤醒的节点Node</span>
<span class="token keyword">for</span> <span class="token punctuation">(</span>Node t <span class="token operator">=</span> tail<span class="token punctuation">;</span> t <span class="token operator">!=</span> null <span class="token operator">&amp;&amp;</span> t <span class="token operator">!=</span> node<span class="token punctuation">;</span> t <span class="token operator">=</span> t<span class="token punctuation">.</span>prev<span class="token punctuation">)</span>
<span class="token keyword">if</span> <span class="token punctuation">(</span>t<span class="token punctuation">.</span>waitStatus <span class="token operator">&lt;=</span> <span class="token number">0</span><span class="token punctuation">)</span>
s <span class="token operator">=</span> t<span class="token punctuation">;</span>
<span class="token punctuation">}</span>
<span class="token comment">//存在需要唤醒的节点，则唤醒它</span>
<span class="token keyword">if</span> <span class="token punctuation">(</span>s <span class="token operator">!=</span> null<span class="token punctuation">)</span>
LockSupport<span class="token punctuation">.</span><span class="token function">unpark</span><span class="token punctuation">(</span>s<span class="token punctuation">.</span>thread<span class="token punctuation">)</span><span class="token punctuation">;</span>

[/code]

}

最上面的判断是清除当前节点的状态，我们重点看下面一部分的逻辑，已经写上了注释，如果下一个节点为null或者为CANCELLED则会从队尾开始找一个可以唤醒的Node进行唤醒。至于为什么从队尾开始寻找，我也不是特别清楚，可能是为了提高一点性能吧（因为如果head的下一个Node状态是CANCELLED，可能它已经等待了很长时间，被用户设置了CANCELLED状态，那么jdk开发人员可能猜测它后面的几个Node的状态可能都是CANCELLED，所以从队尾拿到一个可唤醒的Node遍历的次数可能会少一点）。好了到这里一个Node就已经被唤醒了，这个时候被唤醒的Node会继续执行它的自旋获取锁的逻辑（它阻塞的地方开始继续执行），会继续执行下面的代码的for循环

final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
//继续执行这个自旋，尝试获取锁
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);
}
}

[/code]

代码会走到这里，然后的分析流程就跟上面是一致的…到这里使用AQS队列同步器实现互斥锁（EXCLUSIVE）的逻辑已经全部分析完了，对于共享锁（SHARED）大家可以自行分析，现在接着总结一下AQS实现互斥锁的逻辑

总结：

1、当线程获取锁失败后，会通过CAS加入到同步队列的尾部

2、加入队列的尾部之后，每个队列会做自旋操作，判断前一个Node是不是头节点，如果是则尝试获取锁，否则会进行阻塞，知道它的前一个节点释放锁后唤醒它

3、线程释放锁时会找到它后面的一个可以被唤醒的Node节点，可能从队列head下一个节点，也可能从队尾开始，上面已经说的比较清楚

3、唤醒后的节点会继续从阻塞处进行自行自旋操作，尝试获取锁

本片文章到这里就结束了，希望对大家有点帮助，同时如果哪里写的有问题，欢迎大家指正！

</div><div><div></div></div>
<link href="https://csdnimg.cn/release/phoenix/mdeditor/markdown_views-60ecaf1f42.css" rel="stylesheet">
</div>

内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理

标签：

相关文章推荐

新的分享

章节导航