AQS是并发编程的一个最基本组件,是一个抽象同步器。
网上有很多详细介绍AQS的博文,在这里我就不仔细介绍了,主要写一些重要的内容。
AQS中重要的几个属性:
//同步队列的头节点
private transient volatile Node head;
//同步队列的尾节点
private transient volatile Node tail;
//同步状态
private volatile int state;
由于一个共享资源同一时间可以被一条线程持有,也可以被多个线程持有,因此AQS中存在两种模式,共享模式
和独占模式
。
- 共享模式是共享状态值state每次可以由多个线程持有,如
CountDownLatch
和Semaphore
。 -
独占模式是共享状态值state每次只能由一条线程持有,其他线程如果需要获取,则需要阻塞。如
ReentrantLock
。独占锁的获取
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
先尝试获取锁,获取失败会调用addWaiter
将当前线程添加到同步队列,之后队列中的每个节点会调用acquireQueued()
方法通过自旋的方式先再尝试获取一下锁,如果失败,将当前节点的前驱节点的状态设置为SIGNAL
,并将该线程阻塞,并判断该线程是否被中断。如果被中断了,当前节点获取锁后进行中断操作。
这里用到了模版方法
的设计模式,tryAcquire
是一个抽象方法,具体实现需要到子类中去完成。
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}
下面详细介绍获取独占锁失败后,添加到队列的过程,调用addWaiter()
方法。
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;
}
先创建出一个节点
1) 当前尾节点不为空,采用尾插法利用CAS机制将新创建的节点添加到尾部,设置为尾节点。如果添加失败,就执行enq()方法。
2) 如果尾节点为空,调用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;
}
}
}
}
enq
是一个自旋操作,
1) 如果尾节点为空,说明当前线程是第一个加入同步队列的,先用CAS操作新添一个头节点head
,并将尾节点指向它。第二次循环会跳往另一个执行区域。
2) 利用CAS操作将该节点添加到尾部。
直到自旋添加成功,就结束循环。
入队成功后,就要为该节点开启自旋,尝试获得锁。
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);
}
}
先获取当前节点的先驱节点
1) 如果先驱接节点是头节点并且成功获取同步状态的时候,将当前节点设置为头节点,然后将之前的头节点的next指针设置为null并且pre指针也为null,即将前节点与队列断开,
2)如果获取失败,就调用shouldParkAfterFailedAcquire
方法,主要作用是将该节点的前驱节点的状态设置为SIGNAL
,表示线程阻塞。
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
,就直接返回true;如果状态值大于0,表明该前驱节点已经被取消,则将该前驱节点去除掉,向前移;其他情况,将该前驱节点设置为SIGNAL
。如果添加失败,就返回false,因为是自旋,下一次再尝试。
由于acquireQueued
方法是一个循环,在第二次执行到shouldParkAfterFailedAcquire方法时,由于0号节点的waitStatus已经为Node.SIGNAL
了,所以shouldParkAfterFailedAcquire方法会返回true,然后继续执行parkAndCheckInterrupt
方法,将该线程已经阻塞,并怕判断该线程是否中断。
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
独占锁的释放
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
释放成功后,如果头节点head不为null,并且状态值不为0,就会对它的后继节点进行唤醒。
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)
//后继节点不为null时唤醒该线程
LockSupport.unpark(s.thread);
}
如果头节点的后继节点是空或者它的状态值大于0,表明它是失效的,就要从尾节点节点向前查找,找到最后一个状态值小于等于0的节点,然后对该节点进行唤醒。
共享锁的获取与释放
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
共享锁可以同时被多个线程拥有,可以在初始设置的时候将state
设置为大于0的值,每一个线程获取一次,就减1,当state大于等于0时,别的线程也能够拥有该锁,当小于0时,就不可以,在共享锁模式下,当前线程拿到锁后,会直接通知后继节点去拿锁,而不必等待锁被释放的时候再通知。 在锁释放的时候,支持多个线程释放同步线程同步状态。
参考文章:
深入理解AbstractQueuedSynchronizer(AQS)
java并发编程系列:牛逼的AQS(上)