ReentrantLock源码探究
纳兰小依 人气:1
`ReentrantLock`是一种可重入锁，可重入是说同一个线程可以多次获取同一个锁，内部会有相应的字段记录重入次数，它同时也是一把互斥锁，意味着同时只有一个线程能获取到可重入锁。
## 1.构造函数
```
public ReentrantLock() {
sync = new NonfairSync();
}
public ReentrantLock(boolean fair) {
sync = fair ? new FairSync() : new NonfairSync();
}
```
`ReentrantLock`提供了两个构造函数，构造函数只是用来初始化`sync`字段，可以看到，默认情况下`ReentrantLock`使用的是非公平锁，当然，也可以使用带有布尔参数的构造函数来选择使用公平锁。公平锁和非公平锁的实现依赖于两个内部类：`FairSync`和`NonfairSync`，接下来认识一下这两个类：
```
//非公平锁
static final class NonfairSync extends Sync {
private static final long serialVersionUID = 7316153563782823691L;
/**
* 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);
}
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
}
static final class FairSync extends Sync {
private static final long serialVersionUID = -3000897897090466540L;
final void lock() {
acquire(1);
}
/**
* Fair version of tryAcquire. Don't grant access unless
* recursive call or no waiters or is first.
*/
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
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;
}
}
```
这两个内部类的代码都很短，并且都继承了另一个内部类`Sync`。这里先不急着介绍`Sync`类，因为这个类本身也并不复杂，后续在需要用到其中的方法时顺带讲解，目前只需要知道这个类继承了`AbstractQueuedSynchronizer(AQS)`即可。
## 2.常用方法
- `lock()`
```
public void lock() {
sync.lock();
}
```
`lock`方法提供了加锁的功能，公平锁和非公平锁的加锁操作是不一样的，先来看看非公平锁的细节，接着再讲解公平锁。
- 非公平锁加锁逻辑
```
final void lock() {
//使用CAS操作，尝试将state字段从0修改为1，如果成功修改该字段，则表示获取了互斥锁
//如果获取互斥锁失败，转入acquier()方法逻辑
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
//设置获得了互斥锁的线程
protected final void setExclusiveOwnerThread(Thread thread) {
exclusiveOwnerThread = thread;
}
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
```
对于非公平锁来讲，使用`lock()`方法一上来就尝试获取互斥锁，获取成功就将`exclusiveOwnerThread`指向自己，代表当前是自己持有锁，否则就执行`acquire()`方法的逻辑，下面对`acquire()`方法的逻辑进行逐个分析。
首先是`tryAcquire()`方法，非公平锁重写了该方法，并在内部调用`Sync`类的`nonfairTryAcquire()`。
```
//从上面的逻辑来看，这里的acquires=1
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
//c==0，说明当前处于未加锁状态，锁没有被其他线程获取
if (c == 0) {
//在锁没有被其他线程占有的情况下，非公平锁再次尝试获取锁，获取成功则将exclusiveOwnerThread指向自己
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
//执行到这里说明锁已经被占有，如果是被自己占有，将state字段加1，记录重入次数
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
//当nextc达到int类型最大值时会溢出，因此可重入次数的最大值就是int类型的最大值
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
//执行到这里说明：1）锁未被占有的情况下，抢锁失败，说明当前有其他线程抢到了锁；2）锁已经被其他线程占有
//即只要当前线程没有获取到锁，就返回false
return false;
}
//获取state字段，该字段定义在AQS中
protected final int getState() {
return state;
}
//设置state字段
protected final void setState(int newState) {
state = newState;
}
```
当前线程没有在`tryAcquire()`方法中获取到锁时，会先执行`addWaiter(Node.EXCLUSIVE)`方法，其中参数`Node.EXCLUSIVE`是一个常量，其定义是`static final Node EXCLUSIVE = null`，作用是标记锁的属性是互斥锁。`addWaiter()`方法的作用是将当前线程包装成一个`Node`节点，放入等待队列的队尾，该方法在介绍`CountDownLatch`类时详细讲解过，有兴趣的朋友可以参考[ConcurrentHashMap源码探究（JDK 1.8）](https://www.cnblogs.com/NaLanZiYi-LinEr/p/12391903.html)，本文不再赘述。
将当前线程加入等待队列之后，会接着执行`acquireQueued()`方法，其源码如下：
```
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
//自旋
for (;;) {
//获取当前节点的前一个节点
final Node p = node.predecessor();
//如果前一个节点是头节点，说明当前节点排在队首，非公平锁会则再次通过tryAcquire方法获取锁
if (p == head && tryAcquire(arg)) {
//将自己设置为头节点
setHead(node);
//前一个头结点没用了，会被垃圾回收掉
p.next = null; // help GC
failed = false;
//正常结束，返回false，注意该字段可能会在下面的条件语句中被改变
return interrupted;
}
//如果前一个节点不是头节点，或者当前线程获取锁失败，会执行到这里
//shouldParkAfterFailedAcquire()方法只有在p的状态是SIGNAL时才返回false，此时parkAndCheckInterrupt()方法才有机会执行
//注意外层的自旋，for循环体会一直重试，因此只要执行到这里，总会有机会将p设置成SIGNAL状态从而将当前线程挂起
//另外，如果parkAndCheckInterrupt()返回true，说明当前线程设置了中断状态，会将interrupted设置为true，代码接着自旋，会在上一个条件语句中返回true
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
//如果在自旋中线程被中断或者发送异常，failed字段的值将会为true，这里会处理这种情况，放弃让当前线程获取锁，并抛出中断异常
if (failed)
cancelAcquire(node);
}
}
//方法逻辑是：只有在前置节点的状态是SIGNAL时才返回true，其他情况都返回false
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
return true;
//删除当前节点之前连续状态是CANCELLED的节点
if (ws > 0) {
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
//线程在这里阻塞，并在被唤醒后检查中断状态
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
//
private void cancelAcquire(Node node) {
// Ignore if node doesn't exist
if (node == null)
return;
node.thread = null;
// Skip cancelled predecessors
Node pred = node.prev;
while (pred.waitStatus > 0)
node.prev = pred = pred.prev;
Node predNext = pred.next;
node.waitStatus = Node.CANCELLED;
// If we are the tail, remove ourselves.
if (node == tail && compareAndSetTail(node, pred)) {
compareAndSetNext(pred, predNext, null);
} else {
// If successor needs signal, try to set pred's next-link
// so it will get one. Otherwise wake it up to propagate.
int ws;
if (pred != head &&
((ws = pred.waitStatus) == Node.SIGNAL ||
(ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
pred.thread != null) {
Node next = node.next;
if (next != null && next.waitStatus <= 0)
compareAndSetNext(pred, predNext, next);
} else {
//唤醒后一个节点
unparkSuccessor(node);
}
node.next = node; // help GC
}
}
```
注意`acquireQueued()`要么会抛出中断异常，要么正常结束返回`false`，只有在线程被唤醒后设置了中断状态才会返回`true`。对比可以发现，`acquireQueued()`方法的逻辑与`CountDownLatch`中的`doAcquireSharedInterruptibly()`十分类似，许多方法在`CountDownLatch`这篇博客中讲到过，本文不再对这些方法进行赘述。
介绍完了`acquire()`方法，回过头来看看方法逻辑：
```
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
```
如果在`tryAcquire()`方法中没有获取锁，那么将当前线程加入到等待队列队尾，查看节点的前一个节点是否是头结点，是的话当前线程可以继续向下执行，否则就会阻塞挂起。当`acquireQueued`返回true时，说明线程设置了中断状态，就调用`selfInterrupt()`中断该线程，其他情况`selfInterrupt()`方法没机会执行。
到这里非公平锁的加锁流程已经介绍完了，由于代码逻辑比较长，且看源码的过程中会在好几个类中来回切换，思路很容易断，阅读代码的时候要注意。（有必要补个流程图）。
- 公平锁加锁逻辑
接下来看看公平锁的加锁逻辑：
```
final void lock() {
acquire(1);
}
```
与非公平锁相比，公平锁没有一上来就抢锁的逻辑，这也是公平性的体现。两种锁的`acquire()`方法的框架相同，但是实现细节不同，来看看公平锁的`tryAcquire()`方法：
```
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
//c=0表示当前没有其他线程持有锁
if (c == 0) {
//下面的代码与非公平锁相比，多了hasQueuedPredecessors()方法的处理逻辑，公平锁只有在前面没有其他线程排队的情况下才会尝试获取锁
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;
}
//只要当前线程没有获取到锁，就返回false
return false;
}
public final boolean hasQueuedPredecessors() {
// The correctness of this depends on head being initialized
// before tail and on head.next being accurate if the current
// thread is first in queue.
Node t = tail; // Read fields in reverse initialization order
Node h = head;
Node s;
//h != t表示等待队列中有其他节点
//h.next == null可能会有点费解，按理说h!=t之后，h后面肯定会有节点才对，这种情况其实已经见过，在上文介绍acquireQueued()方法时说过，
//被唤醒的第一个等待节点会将自己设置为头结点，如果这个节点是队列中的唯一节点的话，它的下一个节点就是null
//至于s.thread != Thread.currentThread()这个条件暂时可以忽略，因为公平锁执行到hasQueuedPredecessors方法时根本还没有入队，
//这也意味着，只要队列中有其他节点在等候，公平锁就要求其他线程排队等待
return h != t &&
((s = h.next) == null || s.thread != Thread.currentThread());
}
```
- `lockInterruptibly`
从名字可以看出，`lockInterruptibly`可以响应中断，来看看该方法的实现：
```
public void lockInterruptibly() throws InterruptedException {
sync.acquireInterruptibly(1);
}
public final void acquireInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
//先尝试获取锁，获取失败才执行后面的逻辑
if (!tryAcquire(arg))
doAcquireInterruptibly(arg);
}
private void doAcquireInterruptibly(int arg)
throws InterruptedException {
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;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
```
`lockInterruptibly()`方法几乎与`acquire()`方法完全一样，唯一的区别是`acquire()`方法中，`parkAndCheckInterrupt`因为线程设置了中断状态而返回`true`时，只是简单设置了一下`interrupted`字段的值，而`lockInterruptibly()`则是直接抛出异常。
- `unlock`方法
介绍完加锁的逻辑，接下来看看解锁的逻辑：
```
public void unlock() {
sync.release(1);
}
public final boolean release(int arg) {
//如果成功释放了锁，则执行下面的代码块
if (tryRelease(arg)) {
Node h = head;
//如果头节点不为null，请求节点状态不是初始状态，就释放头结点后第一个有效节点
//问题：这里为什么需要判断头结点的状态呢？？？
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;
//可重入性的判断，如果释放了一次锁，使得c=0，就指针释放锁，做法是将记录锁的字段exclusiveOwnerThread重新指向null
//注意，只有最后一次释放可重入锁，才会返回true
if (c == 0) {
free = true;
setExclusiveOwnerThread(null);
}
setState(c);
return free;
}
//唤醒node节点的下一个有效节点，这里的有效指的是状态不是CANCELLED状态的节点
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
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()`
`ReentrantLock`可以实现绑定多个等待条件，这个功能是在`newCondition()`方法中实现的，每次调用`newCondition()`方法时，都会产生一个新的`ConditionObject`对象，这是`AQS`中的一个内部类，代码很长，这里就不详细讨论了。来简单看看该方法的源码：
```
public Condition newCondition() {
return sync.newCondition();
}
final ConditionObject newCondition() {
return new ConditionObject();
}
```
## 3.总结
在多线程环境中，`ReentrantLock`的非公平锁要比公平锁拥有更高的性能，因为非公平锁避免了线程挂起产生的上下文切换的开销，但是公平锁能够避免线程饥饿问题，因此各有各的使用场景。从源码来看，`J.U.C`包下的很多类都依赖`AQS`类，因此非常有必要搞懂`AQS`。提到`ReentrantLock`，总免不了与`synchronized`进行对比。`synchronized`也是可重入的，并且在`JDK 1.6`以后，`synchronized`的性能已经得到了很大的提升，因此选择使用`ReentrantLock`一般是考虑使用它的三个优势：可中断、可实现公平锁、可绑定多个条件，这些优势是`synchronized`不具备的。
加载全部内容