ReentrantLock、Condition源码分析

ReentrantLock

lock核心原理

1、判断当前锁的状态，如果当前没有人持有锁，则加锁成功
2、如果当前有人持有锁，将当前线程封装为一个Node对象，放入队列（双向链表）
3、挂起当前线程，等待其他线程释放锁时唤醒

加锁源码分析：

• lock方法

final void lock() {
// 利用cas操作修改当前state的值为1，如果操作成功说明没有人持有锁，将当前线程设置为持有锁的线程
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
// 进行加锁操作
acquire(1);
}

• acquire方法

public final void acquire(int arg) {
// 1、tryAcquire(arg)，再次尝试获取锁。
// 2、addWaiter(Node.EXCLUSIVE)，获取锁失败，进行入队操作（双向链表）
// 3、acquireQueued(addWaiter(Node.EXCLUSIVE), arg)，循环当前队列元素，进行线程挂起
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) {
// 对state进行cas
if (compareAndSetState(0, acquires)) {
// 将持有锁的线程设置为当前线程
setExclusiveOwnerThread(current);
return true;
}
}
// 如果当前线程 == 持有锁的线程，说明是可重入锁，cas递增state
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}

公平锁源码

protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
// 如果state==0说明目前没有人持有锁
if (c == 0) {
// hasQueuedPredecessors()，只要队列中存在等待的线程，并且不是当前线程，那么就不能去获取锁
// 如果当前没有其他线程在等待锁，则尝试进行cas操作
if (!hasQueuedPredecessors() &&
compareAndSetState(0, acquires)) {
// 将持有锁的线程设置为当前线程
setExclusiveOwnerThread(current);
return true;
}
}
// 如果当前线程 == 持有锁的线程，说明是可重入锁，cas递增state
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}

• hasQueuedPredecessors方法

// 意思很简单，只要队列中存在等待锁的线程，并且非当前线程，那么就不能去获取锁
public final boolean hasQueuedPredecessors() {
// 头节点
Node t = tail;
// 尾结点
Node h = head;
Node s;
// 如果头节点!=尾结点 &&（头节点的下一个节点 != null || 下一个节点的线程 != 当前线程）
return h != t &&
((s = h.next) == null || s.thread != Thread.currentThread());
}

入队源码

• addWaiter方法

private Node addWaiter(Node mode) {
// 将当前线程封装为一个node
Node node = new Node(Thread.currentThread(), mode);

Node pred = tail;
// 如果tail节点不是空，说明已经有其他节点了，改变指针的指向
if (pred != null) {
// 将当前node的prev指针指向上一个node
node.prev = pred;
// cas操作将当前node设置为tail节点
if (compareAndSetTail(pred, node)) {
// 将上一个node的next指针指向当前node
pred.next = node;
return node;
}
}
// 如果当前node为第一个入队的node，进行初始化操作
enq(node);
return node;
}

• enq方法

private Node enq(final Node node) {
for (;;) {
Node t = tail;
// tail指针 == null说明没有任何node
// 进行初始化操作，构建一个空node，head、tail指向空node
if (t == null) {
if (compareAndSetHead(new Node()))
tail = head;
} else {
// 当前节点的prev指针指向上一个node
node.prev = t;
// cas操作将当前node设置为tail节点
if (compareAndSetTail(t, node)) {
// 将上一个node的next指针指向当前node
t.next = node;
return t;
}
}
}
}

线程挂起源码

• acquireQueued方法

final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
// 获取当前node的上一个node
final Node p = node.predecessor();
// 如果上一个node是head，尝试获取锁
if (p == head && tryAcquire(arg)) {
// 如果获取成功，将当前node设置为head
setHead(node);
// 上一个节点的next指向null，等待GC
p.next = null;
failed = false;
return interrupted;
}
// shouldParkAfterFailedAcquire(p, node)，判断当前线程是否应该挂起
// parkAndCheckInterrupt()，挂起当前线程
// 当其他线程释放了锁，此处对应的线程将会被唤醒，继续下一次循环
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}

• shouldParkAfterFailedAcquire方法

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
// ws == Node.SIGNAL说明应该被挂起
if (ws == Node.SIGNAL)
return true;

// 此判断在加锁中不会被执行
if (ws > 0) {
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
// 将当前node的waitStatus修改为Node.SIGNAL
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}

unlock核心原理

1、当前线程释放锁并且获取队列中下一个node的信息进行唤醒操作

解锁源码：

• unlock方法

public void unlock() {
sync.release(1);
}

• release方法

public final boolean release(int arg) {
// 尝试释放锁
if (tryRelease(arg)) {
// 获取head
Node h = head;
// 如果head不是空，并且waitStatus不是0
if (h != null && h.waitStatus != 0)
// 唤醒下一个node
unparkSuccessor(h);
return true;
}
return false;
}

• tryRelease方法

protected final boolean tryRelease(int releases) {
// 自减state
int c = getState() - releases;
// 如果当前线程不是持有锁的线程直接抛出异常
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
// 如果state自减后为0，说明没有其他人持有锁
if (c == 0) {
free = true;
// 设置当前没有任何线程持有锁
setExclusiveOwnerThread(null);
}
// 重新设置state
setState(c);
return free;
}

• unparkSuccessor方法

private void unparkSuccessor(Node node) {
int ws = node.waitStatus;
// cas重置当前node的waiStatus
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
// 获取下一个node
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;
}
// 唤醒下一个node
if (s != null)
LockSupport.unpark(s.thread);
}

Condition

await核心原理

1、将当前线程封装为一个Node节点加入队列（双向队列，此队列由Condition维护）
2、释放当前线程持有的锁
3、挂起当前线程等待唤醒

• await方法

public final void await() throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
// 添加等待节点
Node node = addConditionWaiter();
// 释放当前线程持有的锁
long savedState = fullyRelease(node);
int interruptMode = 0;
// 检测此节点是否在同步队列上，如果不在，说明此线程还没有资格竞争锁，此线程就继续挂起
while (!isOnSyncQueue(node)) {
// 挂起当前线程
LockSupport.park(this);
// 判断是否中断
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
// 调用的ReentrantLock的方法进行唤醒或者挂起操作
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
// 清理不是处于等待状态的节点
if (node.nextWaiter != null)
unlinkCancelledWaiters();
// 如果不是正常执行（抛了异常），那么执行一些报告相关的工作
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}

signal核心原理

1、将当前等待队列中所有的Node刷入ReentrantLock的等待队列中
2、唤醒await方法中等待队列中挂起的线程

public final void signalAll() {
// 判断当前线程是否是持有锁的线程
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
// 如果队列中存在元素则进行唤醒操作
Node first = firstWaiter;
if (first != null)
doSignalAll(first);
}

private void doSignalAll(Node first) {
// 重置头尾node为空
lastWaiter = firstWaiter = null;
// 循环所有node，进行唤醒操作并将其刷入ReentrantLock的等待队列中
do {
Node next = first.nextWaiter;
first.nextWaiter = null;
transferForSignal(first);
first = next;
} while (first != null);
}

final boolean transferForSignal(Node node) {
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;

// 刷入ReentrantLock
Node p = enq(node);
int ws = p.waitStatus;
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
// 唤醒等待线程
LockSupport.unpark(node.thread);
return true;
}