Java并发包下很多组件的基础都是AQS，例如ReentrantLock，CountDownLatch等等，只要明白了AQS的原理，其他的基本可以理解，那么就从ReentrantLock为例，进行分析
AQS中有个state的状态，以Lock为例，如果哪个线程能成功将state设置为1，那么证明该线程获取到锁
ReentrantLock分为公平锁和非公锁，当时用下面的方式创建的lock是非公平锁
ReentrantLock lock = new ReentrantLock();
那么，先看下ReentrantLock中有什么东西
public class ReentrantLock implements Lock, java.io.Serializable {
private final Sync sync;
abstract static class Sync extends AbstractQueuedSynchronizer {
final boolean nonfairTryAcquire(int acquires) {
....
}
protected final boolean tryRelease(int releases) {
....
}
}
static final class NonfairSync extends Sync {
final void lock() {
....
}
protected final boolean tryAcquire(int acquires) {
....
}
}
static final class FairSync extends Sync {
private static final long serialVersionUID = -3000897897090466540L;
final void lock() {
....
}
protected final boolean tryAcquire(int acquires) {
....
}
}
public ReentrantLock() {
sync = new NonfairSync();
}
public ReentrantLock(boolean fair) {
sync = fair ? new FairSync() : new NonfairSync();
}
}
其中有一个Sync类型的变量，分别有公平锁FairSync和非公平锁NonfairSync的实现，默认是非公平锁，而Sync的又继承于AQS，核心操作都是在这里面的，下面先以NonfairSync的lock方法开始分析
final void lock() {
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
lock的时候，会先调用compareAndSetState方法设置状态(state)值，如果成功，那么将持有锁的线程设置为当前线程
compareAndSetState方法如下：
protected final boolean compareAndSetState(int expect, int update) {
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}
非常粗暴，直接使用CAS操作，当置换成功就代表持有锁，即使是后来的线程，也可能会比先来的线程先持有锁，难怪叫非公平锁
setExclusiveOwnerThread就是简单的将exclusiveOwnerThread变量设置为currentThread
那么获取锁失败会进行什么操作了，如果多个线程都失败了呢，猜一下就是应该有个结构去保存这些失败的线程，当取的锁的线程释放后，这些线程再去竞争，那么acquire就是进行这样的操作
//有个int类型的参数，这个参数在Lock里只是1，而信号量则是设置的信号值，也就是state的值
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
改方法会调用tryAcquire尝试获取锁，该方法由子类实现，非公平锁会调用Sync类的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;
}
- 如果state等于0，表示此刻没有线程获取到锁
2.1 是CAS替换state值，如果成功，设置持有锁的线程，如果失败，代表其他线程在这个时候获取到了 - 如果获取到锁的线程为当前线程，那么将state+1(为什么要加1，是因为Lock是可重入锁，那么就可能lock多次，state就+多次，解锁的时候，需要unlock多次，直到0，才算释放锁)
- 否则，已经有线程持有锁，返回false表示失败
当失败的时候，就会执行下面的代码
acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
那么，我们猜，这里就是保存操作失败的线程数据的地方了，内部是使用了一个链表的结构去保存竞争失败的线程，每个节点的结构就是Node
static final class Node {
static final Node SHARED = new Node();
static final Node EXCLUSIVE = null;
static final int CANCELLED = 1;
static final int SIGNAL = -1;
static final int CONDITION = -2;
static final int PROPAGATE = -3;
volatile int waitStatus;
volatile Node prev;
volatile Node next;
volatile Thread thread;
Node nextWaiter;
}
学过数据结构的对这个结构就很熟悉了，那么节点如何进入链表里的，要看addWaiter方法
private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode);
Node pred = tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
首先会构建一个节点，Node，获取尾指针，由于第一次进行，tail为空，那么会执行enq方法
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;
}
}
}
}
tail为空，那么将一个空节点设置为头节点，且head和tail都指向这个节点，此时结构如下
当链表中没有元素的时候的结构.png
之后，进入第二次循环，t 这时也执行了头节点，且新增的节点的prev执行头节点，执行后就是下面的样子
image.png
再看下addWaiter节点就是往尾部增加节点，节点进入链表后，会调用acquireQueued方法
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
假如此刻有一个线程获取到锁，2个线程进入等待，链表的结构如下
image.png
此时，AB都竞争失败入了链表中，假设B先执行该方法，发现它的前驱(node.predecessor())不是头节点，那么执行shouldParkAfterFailedAcquire方法
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
return true;
// ws>0：状态为CANCELLED
// 找到最前面状态不是CANCELLED的节点，把这之间的节点CANCELLED断开
if (ws > 0) {
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
// 如果前面的节点为正常状态
// 将前面节点的状态设置为SIGNAL
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
该方法返回是否需要挂起线程，看下其中的判断
- 当前节点状态等于SIGNAL，此时需要挂起线程
- 如果状态>0(表示CANCELLED)，那么往前找到一个不是CANCELLED状态的节点，且把CANCELLED状态的节点断开
- 如果上面两个条件不满足，那么将状态设置为SIGNAL然后返回false表示不需要挂起
此时B会重新进入循环判断，前驱仍然不是头节点，但是前驱节点的状态为SIGNAL，那么parkAndCheckInterrupt方法将挂起该线程
再看下A，它的前驱是头节点，那么如果tryAcquire尝试获取锁，如果获取锁的线程还没释放锁，那么和B一样，执行上述流程，如果已经释放了锁，那么tryAcquire返回true，先调用setHead方法
private void setHead(Node node) {
head = node;
node.thread = null;
node.prev = null;
}
将A设置为头节点，Node中的信息置空，看到这里，也就是说将A变成了上面new Node一样的状态，那么我们明白了，头节点这个空节点也就是代表一个持有锁的节点
到这里，A已经获取锁了，可以执行他的代码了，B仍然在挂起状态，直到A释放锁
那么看下释放锁的流程，unLock会调用到AQS的release方法
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
tryRelease在Sync中实现，和tryacquire的过程相反这里就不说了，当tryRelease成功后，调用unparkSuccessor方法唤醒线程
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);
}
- 首先将节点状态设置为0
- 获取该节点后的第一个节点，如果节点不为空且状态不是取消状态，那么唤起该线程
- 如果后续节点为空，或者状态为取消状态，那么从tail往前找到第一个消状态小于等于0的节点
当唤醒成功，即解锁成功，那么刚刚B会从parkAndCheckInterrupt中继续运行，然后重新进入for循环，此时，他的前驱节点是head，且已经释放锁，那么tryacquire也可以成功，之后进行和A一样的操作
看到这里ReentrantLock和AQS的实现就分析完了，上面分析AB的情况只是其中一种简单的情况，在并发的时候，可能顺序并不这样或者节点比较多，分析的情况比较多，不过原理差不多，明白这一种情况后去分析其他的就比较好理解了
注：AQS还有共享锁等的其他实现，这个以后有空会针对某个组件进行分析
网友评论