服务器之家:专注于服务器技术及软件下载分享
分类导航

PHP教程|ASP.NET教程|Java教程|ASP教程|编程技术|正则表达式|C/C++|IOS|C#|Swift|Android|VB|R语言|JavaScript|易语言|vb.net|

服务器之家 - 编程语言 - Java教程 - ReentrantLock实现原理详解

ReentrantLock实现原理详解

2020-08-12 10:56mayday芋头 Java教程

本文将对ReentrantLock实现原理进行详细的介绍,具有很好的参考价值,下面跟着小编一起来看下吧

以下是本篇文章的大纲

1 synchronized和lock

    1.1 synchronized的局限性
    1.2 lock简介

2 aqs

3 lock()与unlock()实现原理

    3.1 基础知识
    3.2 内部结构
    3.3 nonfairsync
    3.3.1 lock()
    3.3.2 unlock()
    3.3.3 小结
    3.4 fairsync

4 超时机制

5 总结

1 synchronized和lock

1.1 synchronized的局限性

synchronized是java内置的关键字,它提供了一种独占的加锁方式。synchronized的获取和释放锁由jvm实现,用户不需要显示的释放锁,非常方便。然而synchronized也有一定的局限性,例如:

当线程尝试获取锁的时候,如果获取不到锁会一直阻塞。

如果获取锁的线程进入休眠或者阻塞,除非当前线程异常,否则其他线程尝试获取锁必须一直等待。

jdk1.5之后发布,加入了doug lea实现的concurrent包。包内提供了lock类,用来提供更多扩展的加锁功能。lock弥补了synchronized的局限,提供了更加细粒度的加锁功能。

1.2 lock简介

lock api如下

?
1
2
3
4
5
6
void lock();
void lockinterruptibly() throws interruptedexception;
boolean trylock();
boolean trylock(long time, timeunit unit) throws interruptedexception;
void unlock();
condition newcondition();

其中最常用的就是lock和unlock操作了。因为使用lock时,需要手动的释放锁,所以需要使用try..catch来包住业务代码,并且在finally中释放锁。典型使用如下

?
1
2
3
4
5
6
7
8
9
10
11
private lock lock = new reentrantlock();
public void test(){
 lock.lock();
 try{
 dosomething();
 }catch (exception e){
 // ignored
 }finally {
 lock.unlock();
 }
}

2 aqs

abstractqueuedsynchronizer简称aqs,是一个用于构建锁和同步容器的框架。事实上concurrent包内许多类都是基于aqs构建,例如reentrantlock,semaphore,countdownlatch,reentrantreadwritelock,futuretask等。aqs解决了在实现同步容器时设计的大量细节问题。

aqs使用一个fifo的队列表示排队等待锁的线程,队列头节点称作“哨兵节点”或者“哑节点”,它不与任何线程关联。其他的节点与等待线程关联,每个节点维护一个等待状态waitstatus。如图

ReentrantLock实现原理详解

aqs中还有一个表示状态的字段state,例如reentrantlocky用它表示线程重入锁的次数,semaphore用它表示剩余的许可数量,futuretask用它表示任务的状态。对state变量值的更新都采用cas操作保证更新操作的原子性。

abstractqueuedsynchronizer继承了abstractownablesynchronizer,这个类只有一个变量:exclusiveownerthread,表示当前占用该锁的线程,并且提供了相应的get,set方法。

理解aqs可以帮助我们更好的理解jcu包中的同步容器。

3 lock()与unlock()实现原理

3.1 基础知识

reentrantlock是lock的默认实现之一。那么lock()和unlock()是怎么实现的呢?首先我们要弄清楚几个概念

可重入锁。可重入锁是指同一个线程可以多次获取同一把锁。reentrantlock和synchronized都是可重入锁。

可中断锁。可中断锁是指线程尝试获取锁的过程中,是否可以响应中断。synchronized是不可中断锁,而reentrantlock则提供了中断功能。

公平锁与非公平锁。公平锁是指多个线程同时尝试获取同一把锁时,获取锁的顺序按照线程达到的顺序,而非公平锁则允许线程“插队”。synchronized是非公平锁,而reentrantlock的默认实现是非公平锁,但是也可以设置为公平锁。

cas操作(compareandswap)。cas操作简单的说就是比较并交换。cas 操作包含三个操作数 —— 内存位置(v)、预期原值(a)和新值(b)。如果内存位置的值与预期原值相匹配,那么处理器会自动将该位置值更新为新值。否则,处理器不做任何操作。无论哪种情况,它都会在 cas 指令之前返回该位置的值。cas 有效地说明了“我认为位置 v 应该包含值 a;如果包含该值,则将 b 放到这个位置;否则,不要更改该位置,只告诉我这个位置现在的值即可。” java并发包(java.util.concurrent)中大量使用了cas操作,涉及到并发的地方都调用了sun.misc.unsafe类方法进行cas操作。

3.2 内部结构

reentrantlock提供了两个构造器,分别是

?
1
2
3
4
5
6
public reentrantlock() {
 sync = new nonfairsync();
}
public reentrantlock(boolean fair) {
 sync = fair ? new fairsync() : new nonfairsync();
}

默认构造器初始化为nonfairsync对象,即非公平锁,而带参数的构造器可以指定使用公平锁和非公平锁。由lock()和unlock的源码可以看到,它们只是分别调用了sync对象的lock()和release(1)方法。

sync是reentrantlock的内部类,它的结构如下

ReentrantLock实现原理详解

可以看到sync扩展了abstractqueuedsynchronizer。

3.3 nonfairsync

我们从源代码出发,分析非公平锁获取锁和释放锁的过程。

3.3.1 lock()

lock()源码如下

?
1
2
3
4
5
6
final void lock() {
 if (compareandsetstate(0, 1))
 setexclusiveownerthread(thread.currentthread());
 else
 acquire(1);
}

首先用一个cas操作,判断state是否是0(表示当前锁未被占用),如果是0则把它置为1,并且设置当前线程为该锁的独占线程,表示获取锁成功。当多个线程同时尝试占用同一个锁时,cas操作只能保证一个线程操作成功,剩下的只能乖乖的去排队啦。

“非公平”即体现在这里,如果占用锁的线程刚释放锁,state置为0,而排队等待锁的线程还未唤醒时,新来的线程就直接抢占了该锁,那么就“插队”了。

若当前有三个线程去竞争锁,假设线程a的cas操作成功了,拿到了锁开开心心的返回了,那么线程b和c则设置state失败,走到了else里面。我们往下看acquire。

acquire(arg)

?
1
2
3
4
5
public final void acquire(int arg) {
 if (!tryacquire(arg) &&
 acquirequeued(addwaiter(node.exclusive), arg))
 selfinterrupt();
}

代码非常简洁,但是背后的逻辑却非常复杂,可见doug lea大神的编程功力。

1. 第一步。尝试去获取锁。如果尝试获取锁成功,方法直接返回。

tryacquire(arg)

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
final boolean nonfairtryacquire(int acquires) {
 //获取当前线程
 final thread current = thread.currentthread();
 //获取state变量值
 int c = getstate();
 if (c == 0) { //没有线程占用锁
 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");
 // 更新state值为新的重入次数
 setstate(nextc);
 return true;
 }
 //获取锁失败
 return false;
}

非公平锁tryacquire的流程是:检查state字段,若为0,表示锁未被占用,那么尝试占用,若不为0,检查当前锁是否被自己占用,若被自己占用,则更新state字段,表示重入锁的次数。如果以上两点都没有成功,则获取锁失败,返回false。

2. 第二步,入队。由于上文中提到线程a已经占用了锁,所以b和c执行tryacquire失败,并且入等待队列。如果线程a拿着锁死死不放,那么b和c就会被挂起。

先看下入队的过程。

先看addwaiter(node.exclusive)

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
/**
 * 将新节点和当前线程关联并且入队列
 * @param mode 独占/共享
 * @return 新节点
 */
private node addwaiter(node mode) {
 //初始化节点,设置关联线程和模式(独占 or 共享)
 node node = new node(thread.currentthread(), mode);
 // 获取尾节点引用
 node pred = tail;
 // 尾节点不为空,说明队列已经初始化过
 if (pred != null) {
 node.prev = pred;
 // 设置新节点为尾节点
 if (compareandsettail(pred, node)) {
 pred.next = node;
 return node;
 }
 }
 // 尾节点为空,说明队列还未初始化,需要初始化head节点并入队新节点
 enq(node);
 return node;
}

b、c线程同时尝试入队列,由于队列尚未初始化,tail==null,故至少会有一个线程会走到enq(node)。我们假设同时走到了enq(node)里。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
/**
 * 初始化队列并且入队新节点
 */
private node enq(final node node) {
 //开始自旋
 for (;;) {
 node t = tail;
 if (t == null) { // must initialize
 // 如果tail为空,则新建一个head节点,并且tail指向head
 if (compareandsethead(new node()))
 tail = head;
 } else {
 node.prev = t;
 // tail不为空,将新节点入队
 if (compareandsettail(t, node)) {
 t.next = node;
 return t;
 }
 }
 }
}

这里体现了经典的自旋+cas组合来实现非阻塞的原子操作。由于compareandsethead的实现使用了unsafe类提供的cas操作,所以只有一个线程会创建head节点成功。假设线程b成功,之后b、c开始第二轮循环,此时tail已经不为空,两个线程都走到else里面。假设b线程compareandsettail成功,那么b就可以返回了,c由于入队失败还需要第三轮循环。最终所有线程都可以成功入队。

当b、c入等待队列后,此时aqs队列如下:

ReentrantLock实现原理详解

3. 第三步,挂起。b和c相继执行acquirequeued(final node node, int arg)。这个方法让已经入队的线程尝试获取锁,若失败则会被挂起。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
/**
 * 已经入队的线程尝试获取锁
 */
final boolean acquirequeued(final node node, int arg) {
 boolean failed = true; //标记是否成功获取锁
 try {
 boolean interrupted = false; //标记线程是否被中断过
 for (;;) {
 final node p = node.predecessor(); //获取前驱节点
 //如果前驱是head,即该结点已成老二,那么便有资格去尝试获取锁
 if (p == head && tryacquire(arg)) {
 sethead(node); // 获取成功,将当前节点设置为head节点
 p.next = null; // 原head节点出队,在某个时间点被gc回收
 failed = false; //获取成功
 return interrupted; //返回是否被中断过
 }
 // 判断获取失败后是否可以挂起,若可以则挂起
 if (shouldparkafterfailedacquire(p, node) &&
  parkandcheckinterrupt())
 // 线程若被中断,设置interrupted为true
 interrupted = true;
 }
 } finally {
 if (failed)
 cancelacquire(node);
 }
}

code里的注释已经很清晰的说明了acquirequeued的执行流程。假设b和c在竞争锁的过程中a一直持有锁,那么它们的tryacquire操作都会失败,因此会走到第2个if语句中。我们再看下shouldparkafterfailedacquire和parkandcheckinterrupt都做了哪些事吧。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
/**
 * 判断当前线程获取锁失败之后是否需要挂起.
 */
private static boolean shouldparkafterfailedacquire(node pred, node node) {
 //前驱节点的状态
 int ws = pred.waitstatus;
 if (ws == node.signal)
 // 前驱节点状态为signal,返回true
 return true;
 // 前驱节点状态为cancelled
 if (ws > 0) {
 // 从队尾向前寻找第一个状态不为cancelled的节点
 do {
 node.prev = pred = pred.prev;
 } while (pred.waitstatus > 0);
 pred.next = node;
 } else {
 // 将前驱节点的状态设置为signal
 compareandsetwaitstatus(pred, ws, node.signal);
 }
 return false;
}
/**
 * 挂起当前线程,返回线程中断状态并重置
 */
private final boolean parkandcheckinterrupt() {
 locksupport.park(this);
 return thread.interrupted();
}

线程入队后能够挂起的前提是,它的前驱节点的状态为signal,它的含义是“hi,前面的兄弟,如果你获取锁并且出队后,记得把我唤醒!”。所以shouldparkafterfailedacquire会先判断当前节点的前驱是否状态符合要求,若符合则返回true,然后调用parkandcheckinterrupt,将自己挂起。如果不符合,再看前驱节点是否>0(cancelled),若是那么向前遍历直到找到第一个符合要求的前驱,若不是则将前驱节点的状态设置为signal。

整个流程中,如果前驱结点的状态不是signal,那么自己就不能安心挂起,需要去找个安心的挂起点,同时可以再尝试下看有没有机会去尝试竞争锁。

最终队列可能会如下图所示

ReentrantLock实现原理详解

线程b和c都已经入队,并且都被挂起。当线程a释放锁的时候,就会去唤醒线程b去获取锁啦。

3.3.2 unlock()

unlock相对于lock就简单很多。源码如下

?
1
2
3
4
5
6
7
8
9
10
11
12
public void unlock() {
 sync.release(1);
}
public final boolean release(int arg) {
 if (tryrelease(arg)) {
 node h = head;
 if (h != null && h.waitstatus != 0)
 unparksuccessor(h);
 return true;
 }
 return false;
}

如果理解了加锁的过程,那么解锁看起来就容易多了。流程大致为先尝试释放锁,若释放成功,那么查看头结点的状态是否为signal,如果是则唤醒头结点的下个节点关联的线程,如果释放失败那么返回false表示解锁失败。这里我们也发现了,每次都只唤起头结点的下一个节点关联的线程。

最后我们再看下tryrelease的执行过程

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
/**
 * 释放当前线程占用的锁
 * @param releases
 * @return 是否释放成功
 */
protected final boolean tryrelease(int releases) {
 // 计算释放后state值
 int c = getstate() - releases;
 // 如果不是当前线程占用锁,那么抛出异常
 if (thread.currentthread() != getexclusiveownerthread())
 throw new illegalmonitorstateexception();
 boolean free = false;
 if (c == 0) {
 // 锁被重入次数为0,表示释放成功
 free = true;
 // 清空独占线程
 setexclusiveownerthread(null);
 }
 // 更新state值
 setstate(c);
 return free;
}

这里入参为1。tryrelease的过程为:当前释放锁的线程若不持有锁,则抛出异常。若持有锁,计算释放后的state值是否为0,若为0表示锁已经被成功释放,并且则清空独占线程,最后更新state值,返回free。

3.3.3 小结

用一张流程图总结一下非公平锁的获取锁的过程。   

ReentrantLock实现原理详解

3.4 fairsync

公平锁和非公平锁不同之处在于,公平锁在获取锁的时候,不会先去检查state状态,而是直接执行aqcuire(1),这里不再赘述。   

4 超时机制

在reetrantlock的trylock(long timeout, timeunit unit) 提供了超时获取锁的功能。它的语义是在指定的时间内如果获取到锁就返回true,获取不到则返回false。这种机制避免了线程无限期的等待锁释放。那么超时的功能是怎么实现的呢?我们还是用非公平锁为例来一探究竟。

?
1
2
3
4
public boolean trylock(long timeout, timeunit unit)
 throws interruptedexception {
 return sync.tryacquirenanos(1, unit.tonanos(timeout));
}

还是调用了内部类里面的方法。我们继续向前探究

?
1
2
3
4
5
6
7
public final boolean tryacquirenanos(int arg, long nanostimeout)
 throws interruptedexception {
 if (thread.interrupted())
 throw new interruptedexception();
 return tryacquire(arg) ||
 doacquirenanos(arg, nanostimeout);
}

这里的语义是:如果线程被中断了,那么直接抛出interruptedexception。如果未中断,先尝试获取锁,获取成功就直接返回,获取失败则进入doacquirenanos。tryacquire我们已经看过,这里重点看一下doacquirenanos做了什么。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
/**
 * 在有限的时间内去竞争锁
 * @return 是否获取成功
 */
private boolean doacquirenanos(int arg, long nanostimeout)
 throws interruptedexception {
 // 起始时间
 long lasttime = system.nanotime();
 // 线程入队
 final node node = addwaiter(node.exclusive);
 boolean failed = true;
 try {
 // 又是自旋!
 for (;;) {
 // 获取前驱节点
 final node p = node.predecessor();
 // 如果前驱是头节点并且占用锁成功,则将当前节点变成头结点
 if (p == head && tryacquire(arg)) {
 sethead(node);
 p.next = null; // help gc
 failed = false;
 return true;
 }
 // 如果已经超时,返回false
 if (nanostimeout <= 0)
 return false;
 // 超时时间未到,且需要挂起
 if (shouldparkafterfailedacquire(p, node) &&
  nanostimeout > spinfortimeoutthreshold)
 // 阻塞当前线程直到超时时间到期
 locksupport.parknanos(this, nanostimeout);
 long now = system.nanotime();
 // 更新nanostimeout
 nanostimeout -= now - lasttime;
 lasttime = now;
 if (thread.interrupted())
 //相应中断
 throw new interruptedexception();
 }
 } finally {
 if (failed)
 cancelacquire(node);
 }
}

doacquirenanos的流程简述为:线程先入等待队列,然后开始自旋,尝试获取锁,获取成功就返回,失败则在队列里找一个安全点把自己挂起直到超时时间过期。这里为什么还需要循环呢?因为当前线程节点的前驱状态可能不是signal,那么在当前这一轮循环中线程不会被挂起,然后更新超时时间,开始新一轮的尝试

5 总结

reentrantlock的核心功能讲解差不多落下帷幕,理解aqs,就很容易理解reentrantlock的实现原理。文中惨杂着笔者的个人理解,如有不正之处,还望指正。

 以上就是本文的全部内容,希望本文的内容对大家的学习或者工作能带来一定的帮助,同时也希望多多支持服务器之家!

原文链接:http://www.cnblogs.com/maypattis/p/6403682.html

延伸 · 阅读

精彩推荐