本文基于jdk1.8进行分析。
reentrantlock是一个可重入锁,在concurrenthashmap中使用了reentrantlock。
首先看一下源码中对reentrantlock的介绍。如下图。reentrantlock是一个可重入的排他锁,它和synchronized的方法和代码有着相同的行为和语义,但有更多的功能。reentrantlock是被最后一个成功lock锁并且还没有unlock的线程拥有着。如果锁没有被别的线程拥有,那么一个线程调用lock方法,就会成功获取锁并返回。如果当前线程已经拥有该锁,那么lock方法会立刻返回。这个可以通过isheldbycurrentthread方法和getholdcount方法进行验证。除了这部分介绍外,类前面的javadoc文档很长,就不在这里全部展开。随着后面介绍源码,会一一涉及到。
1
2
3
4
5
6
7
8
9
10
11
12
|
/** * a reentrant mutual exclusion { @link lock} with the same basic * behavior and semantics as the implicit monitor lock accessed using * { @code synchronized } methods and statements, but with extended * capabilities. * <p>a { @code reentrantlock} is <em>owned</em> by the thread last * successfully locking, but not yet unlocking it. a thread invoking * { @code lock} will return , successfully acquiring the lock, when * the lock is not owned by another thread. the method will return * immediately if the current thread already owns the lock. this can * be checked using methods { @link #isheldbycurrentthread}, and { @link * #getholdcount}. |
首先看一下成员变量,如下图。reentrantlock只有一个成员变量sync,即同步器,这个同步器提供所有的机制。sync是abstractqueuedsynchronizer的子类,同时,sync有2个子类,nonfairsync和fairsync,分别是非公平锁和公平锁。sync,nonfairesync和fairsync的具体实现后面再讲。
1
2
|
/** synchronizer providing all implementation mechanics **/ private final sync sync; |
下面看一下构造函数。如下图。可以看到,reentrantlock默认是非公平锁,它可以通过参数,指定初始化为公平锁或非公平锁。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
/** * creates an instance of {@code reentrantlock}. * this is equivalent to using {@code reentrantlock(false)}. **/ public reentrantlock() { sync = new nonfairsync(); } /** * creates an instance of {@code reentrantlock} with the * given fairness policy. * @param fair {@code true} if this lock should use a fair ordering policy **/ public reentrantlock( boolean fair) { sync = fair ? new fairsync() : new nonfairsync(); } |
下面看一下reentrantlock的主要方法。首先是lock方法。如下图。lock方法的实现很简单,就是调用sync的lock方法。而sync的lock方法是个抽象的,具体实现在nonfairsync和fairsync中。这里我们先不展开讲,而是先读一下lock方法的注释,看看它的作用。lock方法的作用是获取该锁。分为3种情况。
1,如果锁没有被别的线程占有,那么当前线程就可以获取到锁并立刻返回,并把锁计数设置为1。
2,如果当前线程已经占有该锁了,那么就会把锁计数加1,立刻返回。
3,如果锁被另一个线程占有了,那么当前线程就无法再被线程调度,并且开始睡眠,直到获取到锁,在获取到到锁时,会把锁计数设置为1。
lockinterruptibly方法与lock功能类似,但lockinterruptibly方法在等待的过程中,可以响应中断。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
/** * acquires the lock. * <p>acquires the lock if it is not held by another thread and returns * immediately, setting the lock hold count to one. * <p>if the current thread already holds the lock then the hold * count is incremented by one and the method returns immediately. * <p>if the lock is held by another thread then the * current thread becomes disabled for thread scheduling * purposes and lies dormant until the lock has been acquired, * at which time the lock hold count is set to one. **/ public void lock() { sync.lock(); } public void lockinterruptibly() throws interruptedexception { sync.acquireinterruptibly( 1 ); } |
下面,详细看一下非公平锁和公平锁中对lock函数的实现。如下图。下图同时列出了公平锁和非公平锁中lock的实现逻辑。从注释和代码逻辑中,都可以看出,非公平锁进行lock时,先尝试立刻闯入(抢占),如果成功,则获取到锁,如果失败,再执行通常的获取锁的行为,即acquire(1)。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
/** * 非公平锁中的lock * performs lock. try immediate barge, backing up to normal * acquire on failure. **/ final void lock() { if (compareandsetstate( 0 , 1 )) setexclusiveownerthread(thread.currentthread()); else acquire( 1 ); } //公平锁中的lock final void lock() { acquire( 1 ); } |
那么,我们首先了解下,非公平锁“尝试立刻闯入”,究竟做了什么。稍后再继续讲解通常的获取锁的行为。下图是立即闯入行为compareandsetstate(0, 1)的实现。从compareandsetstate函数的注释中,可以知道,如果同步状态值与期望值相等,那么就把它的值设置为updated值。否则同步状态值与期望值不相等,则返回false。这个操作和volatile有着相同的内存语义,也就是说,这个操作对其他线程是可见的。compareandsetstate函数注释里描述的功能,是通过unsafe.compareandswapint方法实现的,而unsafe.compareandswapint是一个native方法,是用c++实现的。那么继续追问,c++底层是怎么实现的?c++底层是通过cas指令来实现的。什么是cas指令呢?来自维基百科的解释是,cas,比较和交换,compare and swap,是用用于实现多线程原子同步的指令。它将内存位置的内容和给定值比较,只有在相同的情况下,将该内存的值设置为新的给定值。这个操作是原子操作。那么继续追问,cas指令的原子性,是如何实现的呢?我们都知道指令时cpu来执行的,在多cpu系统中,内存是共享的,内存和多个cpu都挂在总线上,当一个cpu执行cas指令时,它会先将总线lock位点设置为高电平。如果别的cpu也要执行cas执行,它会发现总线lock位点已经是高电平了,则无法执行cas执行。cpu通过lock保证了指令的原子执行。
现在来看一下非公平锁的lock行为,compareandsetstate(0, 1),它期望锁状态为0,即没有别的线程占用,并把新状态设置为1,即标记为占用状态。如果成功,则非公平锁成功抢到锁,之后setexclusiveownerthread,把自己设置为排他线程。非公平锁这小子太坏了。如果抢占失败,则执行与公平锁相同的操作。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
/** * atomically sets synchronization state to the given updated * value if the current state value equals the expected value. * this operation has memory semantics of a {@code volatile} read * and write. * @param expect the expected value * @param update the new value * @return {@code true} if successful. false return indicates that the actual * value was not equal to the expected value. **/ protected final boolean compareandsetstate( int expect, int update) { // see below for intrinsics setup to support this return unsafe.compareandswapint( this , stateoffset, expect, update); } public final native boolean compareandswapint(object var1, long var2, int var4, int var5); |
下面看一下公平锁获取锁时的行为。如下图。这部分的逻辑有些多,请阅读代码中的注释进行理解。
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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
|
/** * 公平锁的lock **/ final void lock() { acquire( 1 ); } /** * acquires in exclusive mode, ignoring interrupts. implemented * by invoking at least once {@link #tryacquire}, * returning on success. otherwise the thread is queued, possibly * repeatedly blocking and unblocking, invoking {@link * #tryacquire} until success. this method can be used * to implement method {@link lock#lock}. * @param arg the acquire argument. this value is conveyed to * {@link #tryacquire} but is otherwise uninterpreted and * can represent anything you like. **/ public final void acquire( int arg) { /** * acquire首先进行tryacquire()操作。如果tryacquire()成功时则获取到锁,即刻返回。 * 如果tryacquire()false时,会执行acquirequeued(addwaiter(node.exclusive), arg) * 操作。如果acquirequeued(addwaiter(node.exclusive), arg)true时,则当前线程中断自己。 * 如果acquirequeued(addwaiter(node.exclusive), arg)false,则返回。 * 其中tryacquire()操作在nonfairsync中和fairsync中实现又有所区别。 **/ if (!tryacquire(arg) && acquirequeued(addwaiter(node.exclusive), arg)) selfinterrupt(); } /** * nonfairsync中的tryacquire。 * @param acquires * @return **/ protected final boolean tryacquire( int acquires) { return nonfairtryacquire(acquires); } /** * 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(); //首先获取当前同步状态值 int c = getstate(); if (c == 0 ) { //c为0,表示目前没有线程占用锁。没有线程占用锁时,当前线程尝试抢锁,如果抢锁成功,则返回true。 if (compareandsetstate( 0 , acquires)) { setexclusiveownerthread(current); return true ; } } else if (current == getexclusiveownerthread()) { //c不等于0时表示锁被线程占用。如果是当前线程占用了,则将锁计数加上acquires,并返回true。 int nextc = c + acquires; if (nextc < 0 ) // overflow throw new error( "maximum lock count exceeded" ); setstate(nextc); return true ; } //以上情况都不是时,返回false,表示非公平抢锁失败。 return false ; } /** * fair version of tryacquire. don't grant access unless * recursive call or no waiters or is first. * 这个是公平版本的tryacquire **/ protected final boolean tryacquire( int acquires) { final thread current = thread.currentthread(); int c = getstate(); if (c == 0 ) { //c=0时表示锁未被占用。这里是先判断队列中前面是否有别的线程。没有别的线程时,才进行cas操作。 //公平锁之所以公平,正是因为这里。它发现锁未被占用时,首先判断等待队列中是否有别的线程已经在等待了。 //而非公平锁,发现锁未被占用时,根本不管队列中的排队情况,上来就抢。 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 ; } /** * acquires in exclusive uninterruptible mode for thread already in * queue. used by condition wait methods as well as acquire. * 当抢锁失败时,先执行addwaiter(node.exclusive),将当前线程加入等待队列,再执行该方法。 * 该方法的作用是中断当前线程,并进行检查,知道当前线程是队列中的第一个线程,并且抢锁成功时, * 该方法返回。 * @param node the node * @param arg the acquire argument * @return {@code true} if interrupted while waiting **/ 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); } } |
接下来是trylock方法。代码如下。从注释中我们可以理解到,只有当调用trylock时锁没有被别的线程占用,trylock才会获取锁。如果锁没有被另一个线程占用,那么就获取锁,并立刻返回true,并把锁计数设置为1. 甚至在锁被设置为公平排序的情况下,若果锁可用,调用trylock会立刻获取锁,而不管有没有别的线程在等待锁了。从这里我们总结出,不管可重入锁是公平锁还是非公平锁,trylock方法只会是非公平的。
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
|
/** * acquires the lock only if it is not held by another thread at the time * of invocation. * <p>acquires the lock if it is not held by another thread and * returns immediately with the value {@code true}, setting the * lock hold count to one. even when this lock has been set to use a * fair ordering policy, a call to {@code trylock()} <em>will</em> * immediately acquire the lock if it is available, whether or not * other threads are currently waiting for the lock. * this "barging" behavior can be useful in certain * circumstances, even though it breaks fairness. if you want to honor * the fairness setting for this lock, then use * {@link #trylock(long, timeunit) trylock(0, timeunit.seconds) } * which is almost equivalent (it also detects interruption). * <p>if the current thread already holds this lock then the hold * count is incremented by one and the method returns {@code true}. * <p>if the lock is held by another thread then this method will return * immediately with the value {@code false}. * @return {@code true} if the lock was free and was acquired by the * current thread, or the lock was already held by the current * thread; and {@code false} otherwise **/ public boolean trylock() { return sync.nonfairtryacquire( 1 ); } public boolean trylock( long timeout, timeunit unit) throws interruptedexception { return sync.tryacquirenanos( 1 , unit.tonanos(timeout)); } |
接下来是释放锁的方法unlock。代码如下。unlock方式的实现,是以参数1来调用sync.release方法。而release方法是如何实现的呢?release方法首先会调用tryrelease方法,如果tryrelease成功,则唤醒后继者线程。而tryrelease的实现过程十分清晰,首先获取锁状态,锁状态减去参数(放锁次数),得到新状态。然后判断持有锁的线程是否为当前线程,如果不是当前线程,则抛出illegalmonitorstateexception。然后判断,如果新状态为0,说明放锁成功,则把持有锁的线程设置为null,并返回true。如果新状态不为0,则返回false。从tryrelease的返回值来看,它返回的true或false,指的是否成功的释放了该锁。成功的释放该锁的意思是彻底释放锁,别的线程就可以获取锁了。这里要认识到,即便tryrelease返回false,它也只是说明了锁没有完全释放,本次调用的这个释放次数值,依然是释放成功的。
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
|
/** * attempts to release this lock. * <p>if the current thread is the holder of this lock then the hold * count is decremented. if the hold count is now zero then the lock * is released. if the current thread is not the holder of this * lock then {@link illegalmonitorstateexception} is thrown. * @throws illegalmonitorstateexception if the current thread does not * hold this lock **/ public void unlock() { sync.release( 1 ); } /** * releases in exclusive mode. implemented by unblocking one or * more threads if {@link #tryrelease} returns true. * this method can be used to implement method {@link lock#unlock}. * @param arg the release argument. this value is conveyed to * {@link #tryrelease} but is otherwise uninterpreted and * can represent anything you like. * @return the value returned from {@link #tryrelease} **/ public final boolean release( int arg) { if (tryrelease(arg)) { node h = head; if (h != null && h.waitstatus != 0 ) unparksuccessor(h); return true ; } return false ; } 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; } /** * 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 ); /** * 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; 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 ) locksupport.unpark(s.thread); } |
接下来是newcondition方法。关于condition这里不展开介绍,只是了解下该方法的作用。如下图。该方法返回一个和这个锁实例一起使用的condition实例。返回的condition实例支持和object的监控方法例如wait-notify和notifyall相同的用法。
- 1,如果没有获取锁,调用condition的await,signal,signalall方法的任何一个时,会抛出illegalmonitorstateexception异常。
- 2,调用condition的await方法时,锁也会释放,在await返回之前,锁会被重新获取,并且锁计数会恢复到调用await方法时的值。
- 3,如果一个线程在等待的过程中被中断了,那么等待就会结束,并抛出interruptedexception异常,线程的中断标志位会被清理。
- 4,等待的线程以fifo的顺序被唤醒。
- 5,从await方法返回的线程们的获取到锁的顺序,和线程最开始获取锁的顺序相同,这是未指定情况下的默认实现。但是,公平锁更钟爱那些已经等待了最长时间的线程。
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
|
/** * returns a {@link condition} instance for use with this * {@link lock} instance. * <p>the returned {@link condition} instance supports the same * usages as do the {@link object} monitor methods ({@link * object#wait() wait}, {@link object#notify notify}, and {@link * object#notifyall notifyall}) when used with the built-in * monitor lock. * <ul> * <li>if this lock is not held when any of the {@link condition} * {@linkplain condition#await() waiting} or {@linkplain * condition#signal signalling} methods are called, then an {@link * illegalmonitorstateexception} is thrown. * <li>when the condition {@linkplain condition#await() waiting} * methods are called the lock is released and, before they * return, the lock is reacquired and the lock hold count restored * to what it was when the method was called. * <li>if a thread is {@linkplain thread#interrupt interrupted} * while waiting then the wait will terminate, an {@link * interruptedexception} will be thrown, and the thread's * interrupted status will be cleared. * <li> waiting threads are signalled in fifo order. * <li>the ordering of lock reacquisition for threads returning * from waiting methods is the same as for threads initially * acquiring the lock, which is in the default case not specified, * but for <em>fair</em> locks favors those threads that have been * waiting the longest. * </ul> * @return the condition object **/ public condition newcondition() { return sync.newcondition(); } |
可重入锁还有一些其他的方法,这里就不一一介绍了。this is the end.
总结
以上就是这篇文章的全部内容了,希望本文的内容对大家的学习或者工作具有一定的参考学习价值,谢谢大家对服务器之家的支持。如果你想了解更多相关内容请查看下面相关链接
原文链接:https://blog.csdn.net/li_canhui/article/details/85006114