AQS源码泛读，梳理设计流程（jdk8）
2024-09-29 22:35:10
一。AQS简介
AQS（AbstractQueuedSynchronizer）抽象队列同步器，属于多线程编程的基本工具；JDK对其定义得很详细，并提供了多种常用的工具类（重入锁，读写锁，信号量，CyclicBarrier，CountDownLatch），在阅读源码的时候，我是从具体工具类往上读的，这样会比较便于理解AQS的设计。
AQS的核心围绕着一个原子整型数state，通过对它增减控制是否当前线程获得锁，对其的操作使用了CAS的方法。而对于线程的排队机制，使用的是双向链表，每次进入排队的线程会链在最后，同样使用的是CAS添加末尾节点，使用了for(;;)的写法，完整名称为自旋CAS。
下面，我将从五种常用类去分析源码，进而学习AQS。
二。开始吧，重入锁(ReetrantLock)
重入锁的基本概念：它是一种可以将资源独占并且同一个线程可重新获得资源操作权限的锁对象，分为公平锁和非公平锁；公平锁的意义是，按照锁申请的顺序地获得操作权限，非公平锁的意义是，每个线程都是平等的，在每一次调用lock方法的时候均有可能竞争到锁，但如果未获取到锁则仍然会被投入排队队列中按顺序获取。
我们要阅读的重入锁，它首先遵循Lock的规范，并且实现了序列化接口；而Lock的规范，必然定义了如何锁的，如何解锁的，并且规定了newCondition这个方法。而重入锁中，真正使用AQS的是他里面内涵的一个实现类Sync，它继承自AQS，并具有AQS的所有规范。
这个内涵的Sync，在重入锁中实现了两种类型的队列，一个是公平队列，另一个是非公平队列，这取决于你构造重入锁的时候传入的是哪一个，默认是非公平锁；
lock或者trylock这个方法，是进入锁对象逻辑的api，从此方法开始阅读，层层分析，就能明白AQS的机制了。lock方法是AQS里定义的一个抽象方法，被ReentrantLock中的Sync静态类所继承，而Sync类的两种实现：公平锁、非公平锁。重入锁对象的lock方法，实际上是调用内置的Sync对象的lock方法，而Sync的两种实现，在构造重入锁对象的时候指定。
/**
* Acquires the lock.
*
* <p>Acquires the lock if it is not held by another thread and returns
* immediately, setting the lock hold count to one.
*
* <p>If the current thread already holds the lock then the hold
* count is incremented by one and the method returns immediately.
*
* <p>If the lock is held by another thread then the
* current thread becomes disabled for thread scheduling
* purposes and lies dormant until the lock has been acquired,
* at which time the lock hold count is set to one.
*/
public void lock() {
sync.lock();
}
非公平锁申请，通过一个整型数acquires，使用CAS的方式设置AQS中的一个字段state，如果设置成功则将本锁对象的独占线程设置为currentThread；这个else if这里就是重入锁的意义，同一个线程到来的时候，会将state又加上acquires。所以多次调用lock方法的锁对象需要调用同等次数的unlock方法。
/**
* Performs non-fair tryLock. tryAcquire is implemented in
* subclasses, but both need nonfair try for trylock method.
*/
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;
}
公平锁多了一个条件判断，就是hasQueuedPredecessors。就是判断是否为可获得资源的节点，具体为头结点的next节点。
/**
* Fair version of tryAcquire. Don't grant access unless
* recursive call or no waiters or is first.
*/
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
if (!hasQueuedPredecessors() &&
compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
这个方法里面有几种情况会返回false，（h != t）表示等待队列非空，为空就会返回false；另一种情况就是，队列非空，当前队列不等于后继结点的队列，会返回true。
public final boolean hasQueuedPredecessors() {
// The correctness of this depends on head being initialized
// before tail and on head.next being accurate if the current
// thread is first in queue.
Node t = tail; // Read fields in reverse initialization order
Node h = head;
Node s;
return h != t &&
((s = h.next) == null || s.thread != Thread.currentThread());
}
这个方法是独占锁的核心操作，因为它的waiter是EXCLUSIVE。
/**
* Acquires in exclusive mode, ignoring interrupts. Implemented
* by invoking at least once {@link #tryAcquire},
* returning on success. Otherwise the thread is queued, possibly
* repeatedly blocking and unblocking, invoking {@link
* #tryAcquire} until success. This method can be used
* to implement method {@link Lock#lock}.
*
* @param arg the acquire argument. This value is conveyed to
* {@link #tryAcquire} but is otherwise uninterpreted and
* can represent anything you like.
*/
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
这里就是核心的排队代码了，如果当前节点的上一个节点是头结点，也就是队列中的第二个节点，那么就有许可去申请资源了。申请失败，则进入线程park操作。
/*
* Various flavors of acquire, varying in exclusive/shared and
* control modes. Each is mostly the same, but annoyingly
* different. Only a little bit of factoring is possible due to
* interactions of exception mechanics (including ensuring that we
* cancel if tryAcquire throws exception) and other control, at
* least not without hurting performance too much.
*/ /**
* Acquires in exclusive uninterruptible mode for thread already in
* queue. Used by condition wait methods as well as acquire.
*
* @param node the node
* @param arg the acquire argument
* @return {@code true} if interrupted while waiting
*/
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);
}
}
这里有个小地方，就是这个进队方法，如果头节点是空的，会new一个无意义的节点，这就是为什么前面说的第二个节点才是当前可以竞争资源的节点了。
/**
* Inserts node into queue, initializing if necessary. See picture above.
* @param node the node to insert
* @return node's predecessor
*/
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;
}
}
}
}
接着我们去看unlock方法，它指向的是AQS的release接口，与acquire相反，它是将state做减法；
而这个方法，只有是独占线程调用才可以，因为所有lock的非独占线程，全都会被park；
/**
* Releases in exclusive mode. Implemented by unblocking one or
* more threads if {@link #tryRelease} returns true.
* This method can be used to implement method {@link Lock#unlock}.
*
* @param arg the release argument. This value is conveyed to
* {@link #tryRelease} but is otherwise uninterpreted and
* can represent anything you like.
* @return the value returned from {@link #tryRelease}
*/
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
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;
}
主要的唤醒操作在这个方法中执行，它将是传入的head节点的next节点，并且校验出这个得到的next节点是否是想要唤醒的节点。这里判断waitStatus小于0才取用是因为大于0的节点是cancel的节点，需要跳过。在插入队列的时候，是通过CAS设置tail节点的，所以从后往前肯定可以找到想要的节点。（比较难理解）
/**
* Wakes up node's successor, if one exists.
*
* @param node the node
*/
private void unparkSuccessor(Node node) {
/*
* If status is negative (i.e., possibly needing signal) try
* to clear in anticipation of signalling. It is OK if this
* fails or if status is changed by waiting thread.
*/
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0); /*
* Thread to unpark is held in successor, which is normally
* just the next node. But if cancelled or apparently null,
* traverse backwards from tail to find the actual
* non-cancelled successor.
*/
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);
}
三。第二把锁，读写锁(ReetrantReadWriteLock)
读写锁由如下几个成员变量组成，在申请读锁的时候是SHARE模式，申请写锁的时候是独占模式。
他们使用同一个锁对象Sync，在创建对象的时候指定其为公平或者非公平的。读写锁重写了acquireShared方法，非常复杂不易读懂。
/**
* Acquires the write lock.
*
* <p>Acquires the write lock if neither the read nor write lock
* are held by another thread
* and returns immediately, setting the write lock hold count to
* one.
*
* <p>If the current thread already holds the write lock then the
* hold count is incremented by one and the method returns
* immediately.
*
* <p>If the lock is held by another thread then the current
* thread becomes disabled for thread scheduling purposes and
* lies dormant until the write lock has been acquired, at which
* time the write lock hold count is set to one.
*/
public void lock() {
sync.acquire(1);
}
/**
* Acquires the read lock.
*
* <p>Acquires the read lock if the write lock is not held by
* another thread and returns immediately.
*
* <p>If the write lock is held by another thread then
* the current thread becomes disabled for thread scheduling
* purposes and lies dormant until the read lock has been acquired.
*/
public void lock() {
sync.acquireShared(1);
}
四。共享锁，信号量(Semaphore)
信号量是与重入锁完全不同类型的锁，因为他是共享的。它的作用用过的都有印象，就是多个线程可以共享这一把锁，而线程大于初始化凭证之后，就会被阻塞。
/**
* Fair version
*/
static final class FairSync extends Sync {
private static final long serialVersionUID = 2014338818796000944L; FairSync(int permits) {
super(permits);
} protected int tryAcquireShared(int acquires) {
for (;;) {
if (hasQueuedPredecessors())
return -1;
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
}
我相信可以很简单地找到这个FairSync，在semaphore中。它先判断是否只有你来申请，如果不是就回去操作state，如果申请小于0，直接返回，排队。
/**
* Acquires in shared interruptible mode.
* @param arg the acquire argument
*/
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
与重入锁不同的是，入队的时候Node指定为SHARE模式，并且再次尝试获取锁，如果获取的锁是大于等于0的，将会调用setHeadAndPropagate方法传播释放。如果是大于0的，则会调用release接口，下一个唤醒的线程又会重复上述过程，一直唤醒到==0。和重复锁不一样的是，它具有传染性。
/**
* Release action for shared mode -- signals successor and ensures
* propagation. (Note: For exclusive mode, release just amounts
* to calling unparkSuccessor of head if it needs signal.)
*/
private void doReleaseShared() {
/*
* Ensure that a release propagates, even if there are other
* in-progress acquires/releases. This proceeds in the usual
* way of trying to unparkSuccessor of head if it needs
* signal. But if it does not, status is set to PROPAGATE to
* ensure that upon release, propagation continues.
* Additionally, we must loop in case a new node is added
* while we are doing this. Also, unlike other uses of
* unparkSuccessor, we need to know if CAS to reset status
* fails, if so rechecking.
*/
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
unparkSuccessor(h);
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
break;
}
}
只要调用了这个方法，就会进行上述所说的传染；正经的，这个方法就是激活线程并设置PROPAGATE，表示一直往后传播激活。
共享锁和独占锁的区别在于，解锁线程会不会传染...
这样，我们已经读了aqs里面的两种锁了。
五。工具，倒计时器(CountDownLatch)
在看完上面的基本元素之后，搞一个倒计时器是什么鬼，不就是搞个变量然后一直减，然后等于0的时候一股脑儿释放？？？
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
protected boolean tryReleaseShared(int releases) {
// Decrement count; signal when transition to zero
for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
很容易找到上述代码，我们调用倒计时器时候，就会调用countDown方法，每次都会给初始化减一。这时候主线程或者等待线程，调用await在等待。直到它减到0，就会做doReleaseShared，这个时候等待队列只有一个，就是父级线程，它就可以往下走了。因为这货减到0之后不会reset，所以不能复用。
最新文章
- ［Django 2］第一个django应用
- NSUserDefaults存取失败
- Lucene.net
- HTML5+CSS3学习小记
- Python字符进度条
- Red Black Tree in C
- java 操作数据库
- I2C驱动程序框架probe道路
- POJ 2262 Goldbach&#39;s Conjecture(素数相关)
- java 学习笔记 读取配置文件的三种方式
- python中@classmethod @staticmethod区别
- Flask-WTF表单的使用
- 《Linux内核分析》第七周学习总结
- python 读取bin文件
- mfc 类三种继承方式下的访问
- J1850 Implement
- Java实验案例（接口）
- PPI协议（西门子PLCS7-200）
- hbase学习（一）hbase简介
- JAVA中hashmap的分析