Java源码阅读之ReentrantLock - lock和unLock方法

阅读优秀的源码是提升编程技巧的重要手段之一。
如有不对的地方,欢迎指正
转载请注明出处https://blog.lzoro.com

碎碎念

如果需要使用或者了解ReentrantLock,证明已经步入并发编程领域了,这里理论基础不多提,需要的自行查阅资料。

但是,相关术语还是要做一下描述的。

1
2
3
4
5
6
7
8
ReentrantLock:可重入锁

AQS:AbstractQueuedSynchronized 抽象类,队列式同步器

CAS:Compare and Swap, 比较并交换值

CLH队列:The wait queue is a variant of a "CLH" (Craig, Landin, and
* Hagersten) lock queue.

ReentrantLock

首先,贴图大家感受一下。

ReentrantLock

Sync

其中SyncReentrantLock的抽象静态内部类,提供了锁的同步措施,具体实现有NonFairSyncFairSync,分别为公平和非公平锁。

从图中我们可以看出,ReentrantLock是实现了Lock接口和Serializable接口,Serializable是Java的序列化接口,这里我们不多做讨论。

那么,开始源码的阅读了~
首先,先看下Lock接口提供的方法(篇幅所限,这里将源码注释去掉),大致可分为三类:获取锁、释放锁、新建条件(可用于高级应用,如等待/唤醒)。

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
public interface Lock {

/**
* 获取锁,若获取失败则进行等待
*/
void lock();

/**
* 可中断锁
*/
void lockInterruptibly() throws InterruptedException;

/**
* 获取锁,立即返回,成功返回true,否则false
*/
boolean tryLock();

/**
* 获取锁,若获取失败则在指定时间内等待,成功返回true,否则false
*/
boolean tryLock(long time, TimeUnit unit) throws InterruptedException;

/**
* 释放锁
*/
void unlock();

/**
* 新建条件,可用与高级应用
*/
Condition newCondition();
}

接下来我们具体看下ReentrantLock的实现。

1
2
3
public void lock() {
sync.lock();
}

可以看到ReentrantLock的lock方法,是调用静态内部类sysc的lock方法的,而synclock方法是抽象方法,具体的实现有两个,NonfairSync(非公平锁)和FairSync(公平锁),我们先来看NonFairSync的实现

1
2
3
4
5
6
final void lock() {
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}

compareAndSetState(0,1)这个方法是由sysn的父类AbstractQueuedSynchronizer来实现的,也是我们通常说的AQS,而compareAndSetState方法的具体实现是由Unsafe提供的。
Unfafe类的compareAndSwap*系列方法,是虚拟机的本地方法实现,具体的实现不在我们的讨论范围内,简单介绍一下作用,该方法的的作用如下:调用该方法时,若value值与expect值相等,则将value修改为update值,并返回true;若value值与expect值不相等,那么不做任何操作,并返回false,这也就是我们常说的CAS操作,至于存在的ABA等问题和解决方案,有兴趣的可以自己搜索资料。

1
2
3
4
protected final boolean compareAndSetState(int expect, int update) {
// See below for intrinsics setup to support this
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

lock方法执行完CAS操作后

若得的一个true返回,则会执行setExclusiveOwnerThread(Thread.currentThread());,该方法作用是为锁设置独占线程,其实也就是一个赋值操作,如下:

1
2
3
protected final void setExclusiveOwnerThread(Thread thread) {
exclusiveOwnerThread = thread;
}

CAS操作返回一个false,则执行acquire方法

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

从上面源码可以看出,若tryAcquire失败并且acquireQueued返回true中断标识的话,将会中断当前线程。
我们先看一下tryAcquire,方法的作用大致如下:判断锁的state值,若当前未有其他线程持有该锁,则执行CAS操作,成功后则设置独占线程;若发现该锁已被线程持有,则判断持有线程是不是当前线程,若是则允许重入,并判断重入的次数是否超过限制,重入成功后修改state并返回一个true布尔值。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}

final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
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");
setState(nextc);
return true;
}
return false;
}

接下来看一下acquireQueuedaddWaiter方法,作用描述如下,利用addWaiter方法,将当前线程作为一个节点Node加入的CLH队列(The wait queue is a variant of a “CLH” (Craig, Landin, and

* Hagersten) lock queue),这里加入队列有一系列操作,包括尾部判断,前置节点设置等,成功加入队列后,会判断是否有资格去竞争获取锁,有则尝试获取锁,成功后会返回标志位。如果没有资格,则判断是否可以被阻塞,并做相关操作,具体请看注释。
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
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
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,前首节点出队,帮助GC
failed = false;
return interrupted;
}
//判断是否可以阻塞线程并做相应操作,下面具体阅读这几个方法
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
//判断是否获取失败
if (failed)
cancelAcquire(node);
}
}


private Node addWaiter(Node mode) {
//封装成node
Node node = new Node(Thread.currentThread(), mode);
// 获取队列尾节点(作为当前节点的前置节点)
Node pred = tail;
//如果尾节点不为空
if (pred != null) {
//设置当前节点的前置节点
node.prev = pred;
//CAS操作,设置队列尾部
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
//尾节点为null,调用enq方法
enq(node);
//返回当前节点
return node;
}

private Node enq(final Node node) {

for (;;) {
Node t = tail;
//尾节点为null
if (t == null) { // Must initialize
//通过CAS操作设置首节点
if (compareAndSetHead(new Node()))
//将首节点赋值给尾节点(初始化)
tail = head;
} else {
//设置当前节点的前置节点
node.prev = t;
//通过CAS操作设置尾节点
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}

接下来是shouldParkAfterFailedAcquireparkAndCheckInterrupt方法

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
45
46
47
48
49
50
51
52
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
//获取前置节点的等待状态
int ws = pred.waitStatus;
//如果为SIGNAL
if (ws == Node.SIGNAL)
/*
* This node has already set status asking a release
* to signal it, so it can safely park.
*/
//只有当前置节点的状态位SIGNAL的话,当前节点才能进入阻塞,并等待前置节点的唤醒
return true;
if (ws > 0) {
/*
* Predecessor was cancelled. Skip over predecessors and
* indicate retry.
*/
//如果前置节点为取消状态,则不断往前搜索并找到SIGNAL状态的节点,并加在其后面
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.
*/
//通过CAS操作设置前置节点的等待状态位SIGNAL
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}

/**
* 如果上面的方法调用返回true,则代表当前节点可以进入阻塞/等待
*/
private final boolean parkAndCheckInterrupt() {
//通过LockSupport类的park方法来阻塞当前线程
LockSupport.park(this);
//被唤醒后,返回中断标志
return Thread.interrupted();
}

/**
* 这里的阻塞具体实现是JVM虚拟机的本地实现,有兴趣者可以自行研究
*/
public static void park(Object blocker) {
Thread t = Thread.currentThread();
setBlocker(t, blocker);
UNSAFE.park(false, 0L);
setBlocker(t, null);
}

看到这里,会有点困惑,如果没有成功获取到锁的线程进入了阻塞状态,那么它什么时候被唤醒呢?
这里有一个不得不提的点,如果使用lock方法来进行加锁,那么必须成对地使用unlock来释放锁,否则容易导致死锁,一般都是在try-catch-finally进行锁的释放。

所以,等待线程的被唤醒是由持有锁的线程调用unlock后触发的。

接下来,从unlock入手来具体看下源码,可以看到unlock方法是调用sync.release(1)实现的,还是以开头的NonFairSync(非公平锁)的实现来看,

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
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
① 解锁
public void unlock() {
sync.release(1);
}

② 释放锁
public final boolean release(int arg) {
//判断是否释放成功
if (tryRelease(arg)) {
Node h = head;
//判断CLH队列的首节点是否为null,并判断等待状态是否正常
if (h != null && h.waitStatus != 0)
//唤醒节点
unparkSuccessor(h);
return true;
}
return false;
}

③ 释放锁,并唤醒CLH队列中的合法首节点
protected final boolean tryRelease(int releases) {
//计算state和释放数量的差值
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;
}

④ 唤醒阻塞/等待的节点
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);

/*
* 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.
*/
//获取后置节点
Node s = node.next;
//后置节点为null或者为取消状态
if (s == null || s.waitStatus > 0) {
s = null;
//从尾部向前获取到一个不为null且状态不是取消的节点
for (Node t = tail; t != null && t != node; t = t.prev)
if (t.waitStatus <= 0)
s = t;
}
//唤醒该节点
if (s != null)
LockSupport.unpark(s.thread);
}

被唤醒后的节点,返回是否被中断过的标志,在acquireQueued方法内继续执行循环获取锁的流程。

到这里,NonfairSync非公平锁的分析基本上就告一段落了,而关于FairSync的公平机制,有兴趣的可以去阅读下,实现的机制大同小异。

以上,就是Java可重入锁ReentrantLock的lock和unLock源码分析,膜拜Java源码大神。

总结

1、Lock提供locklockInterruptiblytryLock()tryLock(long time, TimeUnit unit)unlocknewCondition五个方法;

2、lockunlock必须成对调用;

3、ReentrantLock实现了Lock和Serializable两个接口;

4、Sync是ReentrantLock的静态内部类,提供了公平锁(FairSync)和非公平锁(NonFairSync)的实现。

5、CAS操作是基于JVM提供的本地方法实现。

6、待补充

卖萌

喜欢的不妨给个赞呗,溜了溜了。

分享到