《Java并发》ReentrantLock实现详解
1. 概述
JDK中独占锁的实现除了使用关键字synchronized外，还可以使用ReenTrantLock。ReenTrantLock和synchronized都是可重入锁，但是ReenTrantLock使用起来比synchronized更为灵活，也更适合复杂的并发场景。
2. 实现原理
ReentrantLock的内部类Sync继承了AQS（AbstractQueuedSynchronizer，队列同步器），先通过CAS尝试获取锁，如果此时已经有线程占据了锁，那么该线程进入 CLH 队列（双向同步队列）。当锁空闲之后，排在CLH队首的线程会被唤醒，然后通过CAS再次尝试获取锁。在这个过程中，需要校验当前锁是公平锁还是非公平锁。
锁的公平与非公平，是指线程请求获取锁的过程中，是否允许插队。
ReentrantLock是以独占锁的加锁策略实现的互斥锁，同时它提供了公平和非公平两种锁获取方式。
- 公平锁：程将按他们发出请求的顺序来获得锁。
- 非公平锁：允许在线程发出请求后立即尝试获取锁，如果可用则可直接获取锁，尝试失败才进行排队等待。
3. 公平锁
3.1 尝试获取锁
// ReentrantLock
final void lock() {
acquire(1);
}
// AQS（AbstractQueuedSynchronizer）
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
// AQS（AbstractQueuedSynchronizer）
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}
线程尝试获取锁时，首先调用ReentranLock.lock()，然后调用AQS.acquire()。但是AQS.acquire()未做具体实现，需要ReentranLock来实现，因为能否成功获取锁是由子类决定。以下是ReentrantLock.tryAcquire()公平锁的具体实现：
// ReentrantLock
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;
}
ReentranLock.tryAcquire()尝试获取锁，该代码块中有两个条件判断：
-
state == 0，则表示当前锁没有被其他线程占用。然后执
hasQueuedPredecessors()判断队首位置是否有等待获取锁的线程，如果队首位置没有等待获取锁的线程，调用compareAndSetState()修改state+1，成功获取锁（多线程同时获取获取锁，需要使用CAS）。最后调用setExclusiveOwnerThread()设置当前线程为独占锁线程。 -
获得锁的线程是否为当前线程，由于ReentrantLock是可重入锁，当已经获取锁的线程每重入一次，state的值递增加1，这也就是重入实现的原理。
注意：如果以上条件都不符合，则返回false，表示锁获取失败，线程进入等待队列。
3.2 线程进入等待队列
private Node addWaiter(Node mode) {
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;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
AQS内部有一条双向同步队列CLH用来存放等待线程，节点是Node。每个Node维护了线程前后指针和等待状态等参数。线程在加入队列前，需要调用addWaiter()封装Node。每个Node需要标记当前是独占锁还是共享锁线程，由传入的mode参数决定。（ReentrantLock 使用的是独占锁模式）
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;
}
}
}
}
创建好Node之后，如果队列不为空，使用CAS将Node加入到队尾。如果加入失败或者队列为空，则调用 enq() 进入死循环，目的是保证Node一定要插入到队列中。当队列为空时，会为头节点创建一个空的 Node。
3.3 阻塞等待线程
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
// 标记1
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
// 标记2
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
线程加入等待队列之后，调用acquireQueued()阻塞线程。
-
标记1是线程唤醒后尝试获取锁的过程，如果前一个节点正好是head，表示自己排在第一位，可以马上调用tryAcquire()尝试获取锁。如果成功获取锁，直接修改自己为head。这一步是实现公平锁的核心，保证锁空闲时，由下个排队线程获取锁。
-
标记2是获取锁失败的处理过程，这个时候，线程可能等着下一次获取，Node变量waitState描述了线程的等待状态，一共四种情况：
- static final int CANCELLED = 1; //取消
- static final int SIGNAL = -1; // 下个节点需要被唤醒
- static final int CONDITION = -2; // 线程在等待条件触发
- static final int PROPAGATE = -3; //（共享锁）状态需要向后传播
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
return true;
if (ws > 0) {
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
shouldParkAfterFailedAcquire()传入当前节点和前一个节点，根据前节点的状态，判断线程是否需要阻塞。
- 前节点状态是
SIGNAL时，当前线程需要阻塞； - 前节点状态是
CANCELLED时，通过循环将当前节点之前所有取消状态的节点移出队列。 - 前节点状态是其他状态时，需要设置前节点为SIGNAL。
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
如果线程需要阻塞，调用parkAndCheckInterrupt()方法进行操作。parkAndCheckInterrupt()使用了LockSupport，和CAS一样，最终使用UNSAFE调用Native方法实现线程阻塞（LockSupport的park和unpark方法作用类似于wait和notify），最后返回线程唤醒后的中断状态。
3.4 释放锁
释放锁的过程：将已获取锁的头节点线程移出队列，然后通知后面等待的节点获取锁。
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;
}
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;
}
因为锁是可以重入的，所以每次lock()会让state+1，对应地每次unlock要让state-1，直到为0时将独占线程变量设置为空，返回标记是否彻底释放锁。最后，调用 unparkSuccessor() 将头节点的下个节点唤醒。
寻找下个待唤醒的线程是从队列尾向前查询的，找到线程后调用LockSupport的unpark()唤醒线程。被唤醒的线程重新执行acquireQueued里的循环，就是上文关于acquireQueued标记1部分，线程重新尝试获取锁。
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);
}
3.5 中断锁
ReentrantLock中的lockInterruptibly()方法使得线程可以在被阻塞时响应中断，比如一个线程t1通过lockInterruptibly()方法获取到一个可重入锁，并执行一个长时间的任务，另一个线程通过interrupt()方法就可以立刻打断t1线程的执行，来获取t1持有的那个可重入锁。而通过ReentrantLock的lock()方法或者Synchronized持有锁的线程是不会响应其他线程的interrupt()方法的，直到该方法主动释放锁之后才会响应interrupt()方法。
在尝试获取锁过程中，acquire()里最后一行代码调用了selfInterrupt()，对当前线程发送一个中断请求。之所以要这样操作，其实时因为LockSupport的park()阻塞线程后，有两种情况可能被唤醒：
-
第一种情况：头节点释放锁后，会调用
LockSupport.unpark()唤醒下个节点的线程。整个过程没有涉及到中断，最终acquireQueued返回 false 时，不需要调用selfInterrupt()。 -
第二种情况：
LockSupport.park()支持响应中断请求，能被其他线程通过interrupt()唤醒。但这种唤醒没有用，因为线程前面可能还有等待线程，在acquireQueued的循环里，线程会再次被阻塞。parkAndCheckInterrupt()返回的是Thread.interruptd()，不仅返回中断状态，还会清除状态，保证阻塞线程忽略中断。最终acquireQueued返回true时，真正的中断状态已经被清除，需要调用selfInterrupt维持中断状态。
因此普通的lock()方法并不能被其他线程中断，ReentrantLock时可以支持中断，需要使用lockInterruptibly。
4. 非公平锁
公平锁和非公平锁的区别主要是获取锁的过程不同。在非公平锁的lock()方法里，第一步直接尝试将state修改为1，很明显，这时抢先获取锁的过程。如果state修改失败，则和公平锁一样，调用acquire。
final void lock() {
if (compareAndSetState(0, 1)) // 如果一开始未上锁，直接抢占锁
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
// 线程尝试获取锁的时候，首先调用 nonfairTryAcquire()
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;
}
nonfairTryAcquire()和tryAcquire()的差异仅仅是缺少调用hasQueuedPredecessors()，这一点体现出公平锁和非公平锁的不同，公平锁会关注队列里的排队情况，老老实实按照FIFO的次序，而非公平锁只要有机会就抢占，不关心排队情况，谁出手快谁就先获得锁。
5. ReentrantLock和Condition配合使用实现等待/通知机制
关键词synchronized与wait()、notify()、notifyAll()结合可以实现等待/通知机制。而ReentrantLock和Condition配合也可以实现等待/通知机制。
- 使用notify()和notifyAll()无法选择性的通知线程。
- 使用
Condition可以在Lock对象里面创建多个Condition（对象监视器）实例，线程对象可以注册在指定的Condition中，从而实现有选择性地进行线程通知，使线程调度更加的灵活。
public class ReentrantLockCondition {
private Lock lock = new ReentrantLock();
private Condition condition = lock.newCondition();
private static ExecutorService executorService = Executors.newFixedThreadPool(20);
private static ReentrantLockCondition reentrantLockCondition = new ReentrantLockCondition();
public static void main(String[] args) throws Exception {
// executorService.execute(new MyThread());
// Thread.sleep(5000);
// reentrantLockCondition.signal();
for (int i = 1; i <= 20; i++) {
executorService.execute(new MyThread());
Thread.sleep(1000);
reentrantLockCondition.signalAll();
}
}
/**
* 等待
*/
public void await() {
try {
lock.lock();
System.out.println("await time:" + System.currentTimeMillis());
condition.await(); // 使线程处于等待状态
} catch (InterruptedException ex) {// 线程中断异常
ex.printStackTrace();
} finally {
lock.unlock();
}
}
/**
* 通知
*/
public void signal() {
try {
lock.lock();
System.out.println("signal time:" + System.currentTimeMillis());
condition.signal();// 通知等待线程进行唤醒
} catch (IllegalMonitorStateException ex) {// 监视器不合法异常
ex.printStackTrace();
} finally {
lock.unlock();
}
}
/**
* 通知
*/
public void signalAll() {
try {
lock.lock();
System.out.println("signal time:" + System.currentTimeMillis());
condition.signalAll();// 通知等待线程进行唤醒
} catch (IllegalMonitorStateException ex) {// 监视器不合法异常
ex.printStackTrace();
} finally {
lock.unlock();
}
}
static class MyThread implements Runnable {
@Override
public void run() {
reentrantLockCondition.await();
}
}
}
6. ReentrantLock和synchronized的区别
-
锁可重入
ReentrantLock和syncronized关键字一样，都是可重入锁。RetrantLock利用AQS的的state状态来判断资源是否已锁，同一线程重入加锁， state的状态 +1 ; 同一线程重入解锁，state状态-1 (解锁必须为当前独占线程，否则异常)；当state为0时解锁成功。 -
手动加锁和释放锁
synchronized关键字是自动进行加锁、解锁的；而ReentrantLock需要lock()和unlock()方法配合try/finally语句块来完成，来手动加锁、解锁。 -
支持设置锁的超时时间
synchronized关键字无法设置锁的超时时间，如果一个获得锁的线程内部发生死锁，那么其他线程就会一直进入阻塞状态；而ReentrantLock提供tryLock方法，允许设置线程获取锁的超时时间，如果超时，则跳过，不进行任何操作，避免死锁的发生。 -
支持公平/非公平锁
synchronized关键字是一种非公平锁，先抢到锁的线程先执行。而ReentrantLock的构造方法中允许设置true/false来实现公平、非公平锁，如果设置为true ，则线程获取锁要遵循"先来后到"的规则，每次都会构造一个线程Node ，然后插入到双向链表尾部排队，等待前面的Node释放锁资源。 -
锁可中断
ReentrantLock中的lockInterruptibly()方法使得线程可以在被阻塞时响应中断，例如一个线程T1通过lockInterruptibly()方法获取到一个可重入锁，并执行一个长时间的任务，另一个线程通过interrupt()方法就可以立刻打断T1线程的执行，来获取T1持有的那个可重入锁。而通过ReentrantLock的lock()方法或者synchronized持有锁的线程是不会响应其他线程的interrupt()方法的，直到该方法主动释放锁之后才会响应interrupt()方法。
