今天我们来研究学习一下ReentrantLock类的相关原理，ReentrantLock的内部使用AbstractQueuedSynchronizer实现线程锁。除了ReentrantLock在java.util.concurrent包中还有很多类都依赖于这个类所提供队列式同步器。
为了方便学习我们以ReentrantLock为例，来学习ReentrantLock和AbstractQueuedSynchronizer。
本节我们先学习ReentrantLock是如何加锁的。ReentrantLock.lock()内部调用sync.lock()
ReentrantLock的内部有两种实现，FairSync（公平锁）和NonfairSync（非公平锁）都继承了Sync类，
而Sync类继承了AbstractQueuedSynchronizer。
AbstractQueuedSynchronizer
AbstractQueuedSynchronizer里有四个可以重写的方法，可以用它们实现不同的功能。例如ReentrantLock、CountDownLatch、Semaphore等。
//独占模式下获取许可
@Override
protected boolean tryAcquire(int arg) {
return super.tryAcquire(arg);
}
//独占模式下释放许可
@Override
protected boolean tryRelease(int arg) {
return super.tryRelease(arg);
}
//共享模式下获取许可
@Override
protected int tryAcquireShared(int arg) {
return super.tryAcquireShared(arg);
}
//共享模式下释放许可
@Override
protected boolean tryReleaseShared(int arg) {
return super.tryReleaseShared(arg);
}
lock操作
下面将以NonfairSync为例，讲解ReentrantLock是如何加锁。
AbstractQueuedSynchronizer中有一个属性state，为0时表示没有线程持有锁，大于0时表示有锁。ReentrantLock是可重入表，因此state用于重入计数。
public class ReentrantLock {
private Sync sync;
//加锁
public void lock() {
sync.lock();
}
static final class NonfairSync extends Sync {
final void lock() {
//CAS方式更新state变量的值，从0更新到1成功表示获取到锁
if (compareAndSetState(0, 1))
//记录获得锁的线程
setExclusiveOwnerThread(Thread.currentThread());
else
//其他线程已经获取锁走acquire
acquire(1);
}
}
abstract static class Sync extends AbstractQueuedSynchronizer {
......
}
}
acquire
acquire函数的作用是获取同一时间段内只能被一个线程获取的许可。
先调用tryAcquire尝试获取一次锁，如果失败则调用acquireQueued等待锁。
public final void acquire(int arg) {
//tryAcquire尝试获取锁
//但是多线程抢锁可能当前线程抢不到，走acquireQueued
//addWaiter为当前线程创建一个WaitNode节点加入队列（链表）
if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) {
selfInterrupt();
}
}
//尝试获取一次锁
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
//state为0表示没有人持有锁，可以尝试获取锁
if (c == 0) {
//CAS方式更新state变量的值，从0更新到1成功表示获取到锁
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;
}
addWaiter
AbstractQueuedSynchronizer中实现了链表，用于保存所有的等待锁的线程调用addWaiter将当前线程封装成链表中的一个节点加入链表中。
//链表头节点
private transient volatile Node head;
//链表尾节点
private transient volatile Node tail;
//addWaiter为当前线程创建一个WaitNode节点加入队列（链表）
private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode);
//尝试向链表加入节点
//第一次调用addWaiter时tail为null，直接走enq
Node pred = tail;
if (pred != null) {
node.prev = pred;
//CAS方式插入node
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
private Node enq(final Node node) {
//不停尝试插入node
for (;;) {
Node t = tail;
if (t == null) {
if (compareAndSetHead(new Node()))
tail = head;
} else {
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
acquireQueued
addWaiter成功后，调用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;
return interrupted;
}
if (//检查node状态
shouldParkAfterFailedAcquire(p, node)
//阻塞线程
//如果parkAndCheckInterrupt==true时线程已经中断
&& parkAndCheckInterrupt()) {
interrupted = true;
}
}
} finally {
if (failed)
cancelAcquire(node);
}
}
park
如果其他线程持有锁，那么让当前线程阻塞。等到其他线程释放锁时唤醒线程
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) {
//检查前驱节点waitStatus（初始化时值为0）
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
//如果前继的节点状态为SIGNAL，表示当前节点需要阻塞
return true;
if (ws > 0) {
//过滤掉所有cancelled的节点
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);
//返回false，表示线程没有中断
return Thread.interrupted();
}
unlock操作
//解锁
public void unlock() {
sync.release(1);
}
public final boolean release(int arg) {
//尝试释放一个许可
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
//唤醒下一个Node中的线程
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;
}
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);
}
