ReentrantLock
互斥锁,可重入
AQS是可以支持互斥锁和共享锁的，这里只分析互斥锁的源码
加锁
公平锁和非公平锁
- 公平锁
final void lock() {
acquire(1); //抢占1把锁.
}
// AQS里面的方法
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) { //表示无锁状态
if (!hasQueuedPredecessors() &&
compareAndSetState(0, acquires)) { //CAS原子操作
setExclusiveOwnerThread(current); //把获得锁的线程保存到exclusiveOwnerThread中
return true;
}
}
//如果当前获得锁的线程和当前抢占锁的线程是同一个，表示重入
else if (current == getExclusiveOwnerThread()) {
//增加重入次数.
int nextc = c + acquires;
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
setState(nextc); //保存state
return true;
}
return false;
}
- 非公平锁
final void lock() {
//非公平锁,不管当前AQS队列中是否已有线程在排队的，都先去插队(尝试获得锁)
if (compareAndSetState(0, 1)) //返回false表示抢占锁失败
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
//AQS里的方法
public final void acquire(int 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();
if (c == 0) {
//hasQueuedPredecessors 和公平锁的区别，公平锁是如果已经有在排队的线程了，
// 那么新过来的线程就不允许插队(去尝试获取锁)
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;
}
- 加入队列并进行自旋等待
public final void acquire(int arg) {
//如果尝试获取锁失败，则会加入队列并进行自旋等待
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
- addWaiter(Node.EXCLUSIVE) -> 添加一个互斥锁的节点
- acquireQueued() -> 自旋锁和阻塞的操作
private Node addWaiter(Node mode) {
//把当前线程封装成一个Node节点。后续唤醒线程的时候，需要得到被唤醒的线程.
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
Node pred = tail;
//假设不存在竞争,那么一次CAS操作就可以将Node节点加入到双向链表，如果存在竞争，最终还是要通过enq里的自旋来加入到双向链表中
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
//第一次添加Waiter的时候pred一定为空,会直接执行enq方法
enq(node);
return node;
}
//从尾部添加到链表，尾插法
private Node enq(final Node node) {
//自旋
for (;;) {
Node t = tail;
if (t == null) { // Must initialize
//初始化一个head节点,注意Head节点是一个空节点,不存储任何线程信息
if (compareAndSetHead(new Node()))
tail = head;
} else {
//注意这里先设置的新节点node的prev,所以后续在遍历链表的时候都是从tail->head
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
//node表示当前来抢占锁的线程节点，可能是Thread B,Thread C
final boolean acquireQueued(final Node node, int arg) {
//标记是否成功拿到锁
boolean failed = true;
try {
//标记等待过程中是否中断过
boolean interrupted = false;
for (;;) { //自旋
//begin
//获取当前节点的前置节点,如果是head(则表示这个线程是排在第一个),会尝试去获取锁
//tryAcquire 公平锁里判断当前节点是否有前置节点来决定能否去抢占锁
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
//拿到资源后，将head指向该结点。
setHead(node);
//setHead中node.prev已置为null，此处再将head.next置为null，就是为了方便GC回收以前的head结点。
p.next = null;
failed = false;
//返回等待过程中是否被中断过
return interrupted;
}
//end begin->end 这里会尝试去获得锁，如果失败的话会让线程去阻塞(park),如果设置前驱节点节点的signal状态失败,那么会再次尝试去获取锁,失败后再次尝试设置(自旋)
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt()) //LockSupport.park
//如果等待过程中被中断过，就将interrupted标记为true，注意此时的failed=true,因为是通过中断唤醒的,并没有获取到锁资源
interrupted = true;
}
} finally {
//如果等待过程中没有成功获取资源（如timeout，或者可中断的情况下被中断了），那么取消结点在队列中的等待。
if (failed)
cancelAcquire(node);
}
}
先来看下shouldParkAfterFailedAcquire和parkAndCheckInterrupt()的实现再来看acquireQueued的代码，Node节点的状态有如下几种:
- SIGNAL: -1 表示它的下一个节点处于park状态，如果当前节点取消或者释放锁资源的时候需要unpark下一个节点
- CANCELLED: 1 :当前节点由于超时或者被中断而处于取消状态
- CONDITION：-2 共享锁使用的状态
- PROPAGATE: -3
默认为0
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
//前驱节点已经是SIGNAL状态了,当前节点可以安心去park
return true;
if (ws > 0) {
//如果前驱节点处于CANCELLED状态,那么就需要遍历链表(tail->head),直到找到最近一个正常状态的节点,排在其后
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
//如果前驱节点非CANCELLED状态,那就把前驱节点的状态设置成SIGNAL(告诉他等他释放锁资源需要通知自己) 并发场景下CAS可能会失败(失败后会在外层方法自旋,再次触发该方法)
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
这里之所以通过 node.prev = pred = pred.prev; 从后往前遍历,去掉为CANCELLED状态的节点，因为我们在设置双向链表的时候是先设置的tail的prev节点,如果从前往后操作，并发场景下可能会出现next指针未指向的情形,导致异常
在整个过程中，如果前驱结点的状态不是SIGNAL，那么自己就不能安心去park，需要去找个安心的休息点，同时可以再尝试下看有没有机会获取锁资源。
//ThreadB、 ThreadC -> 都会阻塞在下面这个代码的位置.
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this); //调用park()使线程进入waiting状态
return Thread.interrupted(); //如果被唤醒，查看自己是不是被中断唤醒的。线程除了被正常唤醒之外,interrupt方法也会唤醒线程 Thread.interrupted()判断当前线程是否被中断过
}
再来总结下acquireQueued()的流程：
node进入队尾后
1.先判断前驱节点是否为head,如果是则尝试去获取锁资源，拿到锁资源后就将head指向当前节点
2. 获取不到锁资源则检查前驱节点的状态，找到安全休息点(前驱节点为SIGNAL状态)则调用park进入等待状态
3.被唤醒后，先判断从入队到唤醒过程中是否有被中断过，如果有的话则将本节点置为CANCELLED状态，否则尝试去获取锁资源，执行步骤1。
解锁
public final boolean release(int arg) {
if (tryRelease(arg)) {
//得到当前AQS队列中的head节点
Node h = head;
//head节点不为空且状态不是0,说明这个节点不是链表里的最后一个节点
if (h != null && h.waitStatus != 0)
//唤醒下一个节点(在里面会调用unpark方法)
unparkSuccessor(h);
return true;
}
return false;
}
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;
if (ws < 0) //SIGNAL状态,表示为可以唤醒状态
compareAndSetWaitStatus(node, ws, 0); //恢复成0
//查找下一个需要唤醒的结点s
Node s = node.next;
//说明ThreadB这个线程可能已经被销毁，或者出现异常...
if (s == null || s.waitStatus > 0) {
s = null;
//从tail -> head进行遍历,进行节点的移除
for (Node t = tail; t != null && t != node; t = t.prev)
//查找到小于等于0的节点(我们之前修改节点的状态是将前驱节点设置为SIGNAL,所以最后一个节点的状态是默认的0)
if (t.waitStatus <= 0)
s = t;
}
if (s != null)
//唤醒封装在Node中的被阻塞的线程)
LockSupport.unpark(s.thread);
}
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;
}
