AbstractQueuedSynchronizer源码解读 AbstractQueuedSynchronizer源码解读

?? AbstractQueuedSynchronizer（AQS），是 Java 并发包中，实现各种同步结构和部分其他组成单元（如线程池中的 Worker）的基础。AQS 内部数据和方法，可以简单拆分为：

一个 volatile 的int成员表征状态(state属性)，同时提供了 setState 和 getState 方法。这里之所以用int不用boolean是因为有场景并不是只能有一个线程获取锁，比如CountDownLatch就允许多个线程获取锁。
1： ReentrantLock: 用state变量记录某个线程获取独占锁的次数,获取锁时+1，释放锁时-1，在获取时会校验线程是否可以获取锁。
2：Semaphore: 用state变量作为计数器，只有在大于0时允许线程进入。获取锁时-1，释放锁时+1。
3： CountDownLatch: 用state变量作为计数器，在初始化时指定。只要state还大于0，获取共享锁会因为失败而阻塞，直到计数器的值为0时，共享锁才允许获取，所有等待线程会被逐一唤醒。
一个先入先出（FIFO）的等待线程队列，以实现多线程间竞争和等待，这是 AQS 机制的核心之一。
各种基于 CAS 的基础操作方法，以及各种期望具体同步结构去实现的 tryAcquire/tryRelease 方法。

文章图片
AbstractQueuedSynchronizer关键属性

下面来看看acquire获取锁的部分源码：

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { //定义为final说明子类不能覆盖这个实现过程，之类只需要实现tryAcquire方法 public final void acquire(int arg) { //这里分别会调用tryAcquire方法用于获取锁，如果失败，会调用addWaiter方法，将当前线的Node加入到FIFO队列里，通过acquireQueued判断当前节点是否为头节点，如果是则试图获取锁，获取不成功会加入等待队列，并把线程挂起，不是头结点的话直接加入等待队列并挂起线程。 if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }//这个方法在AbstractQueuedSynchronizer里留给子类来实现， protected boolean tryAcquire(int arg) { throw new UnsupportedOperationException(); }}

这里分别会调用tryAcquire方法用于获取锁，如果失败，会调用addWaiter方法，将当前线的Node加入到FIFO队列里，通过acquireQueued判断当前节点是否为头节点，如果是则试图获取锁，获取不成功会加入等待队列，并把线程挂起，不是头结点的话直接加入等待队列并挂起线程。在AbstractQueuedSynchronizer里tryAcquire方法是留给子类来实现的，下面来看看ReentrantLock里的非公平锁是如何实现这里的逻辑的。源码如下：

abstract static class Sync extends AbstractQueuedSynchronizer { //非公平锁最终最调用到当前这个方法，传入的acquires为1 final boolean nonfairTryAcquire(int acquires) { final Thread current = Thread.currentThread(); int c = getState(); // 获取当前 AQS 内部状态量 if (c == 0) { // 0 表示无人占有，则直接用 CAS 修改状态位， if (compareAndSetState(0, acquires)) {// 不检查排队情况，直接争抢 setExclusiveOwnerThread(current); // 并设置当前线程独占锁 return true; } } else if (current == getExclusiveOwnerThread()) { // 即使状态不是 0，也可能当前线程是锁持有者，因为这是重入锁 int nextc = c + acquires; if (nextc < 0) // overflow throw new Error("Maximum lock count exceeded"); setState(nextc); return true; } return false; }}

ReentrantLock的公平锁的实现逻辑如下：

static final class FairSync extends Sync { //公平锁的实现逻辑 protected final boolean tryAcquire(int acquires) { final Thread current = Thread.currentThread(); int c = getState(); // 获取当前 AQS 内部状态量 if (c == 0) {// 0 表示无人占有， if (!hasQueuedPredecessors() && //这里需要判断队列头节点的的下一个节点为空。也就是说不允许线程进行插队。 compareAndSetState(0, acquires)) { //为空才会尝试获取锁 setExclusiveOwnerThread(current); return true; } } else if (current == getExclusiveOwnerThread()) { // 即使状态不是 0，也可能当前线程是锁持有者，因为这是重入锁 int nextc = c + acquires; if (nextc < 0) throw new Error("Maximum lock count exceeded"); setState(nextc); return true; } return false; } }

接下来再分析 acquireQueued，如果前面的 tryAcquire 失败，代表着锁争抢失败，进入排队竞争阶段。这里就是利用 FIFO 队列，实现线程间对锁的竞争的部分，算是AQS 的核心逻辑。源码如下：

//这个方法是不允许子类重写的 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)) { / 如果前一个节点是头结点，则当前节点是FIFO队列里的第一个节点，则会再次调用tryAcquire方法尝试获取锁。 setHead(node); //成功，则设置当前节点为头节点 p.next = null; // help GC 清空前面节点对当前节点的引用 failed = false; return interrupted; } if (shouldParkAfterFailedAcquire(p, node) && //获取锁失败后，需要判断当前线程是否不要挂起，正常情况下是都会需要挂起的 parkAndCheckInterrupt()) interrupted = true; //这里需要设置一个标识位，是因为park方法无法感知到其他线程调用了当前线程的interrupt方法改变了现场的interrupt状态 } } finally { if (failed) // 获取锁失败，调用cancelAcquire方法 cancelAcquire(node); } }

【AbstractQueuedSynchronizer源码解读】acquireQueued 的逻辑，简要来说，就是如果当前节点的前面是头节点（AQS里的头结点是哨兵结点），则试图获取锁，一切顺利则成为新的头节点；否则，会挂起线程。上面的部分分析了获取锁的逻辑。下面再来看看释放锁的逻辑AbstractQueuedSynchronizer的relase源码如下：

//释放锁需要调用release方法 public final boolean release(int arg) { if (tryRelease(arg)) { //释放成功后，调用unparkSuccessor方法，让下个节点得到锁 Node h = head; if (h != null && h.waitStatus != 0) unparkSuccessor(h); return true; }//留给具体的子类来实现 protected boolean tryRelease(int arg) { throw new UnsupportedOperationException(); }//当前节点释放锁，下个节点得到锁状态 private void unparkSuccessor(Node node) { int ws = node.waitStatus; //设置当前节点的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; } //下一个节点不为空，则调用LockSupport的unpark方法唤醒线程 if (s != null) LockSupport.unpark(s.thread); }

ReentrantLock里tryRelease的实现如下所示：

protected final boolean tryRelease(int releases) { int c = getState() - releases; //用state减去release，这个状态是新的状态 if (Thread.currentThread() != getExclusiveOwnerThread()) throw new IllegalMonitorStateException(); boolean free = false; //释放的结果用free来表示 if (c == 0) { //为零，表求当前没有钱程点用了， free = true; setExclusiveOwnerThread(null); } //设置状态 setState(c); return free; }

Node的状态 java.util.concurrent.locks.AbstractQueuedSynchronizer.Node

/** 表示线程已取消 */ static final int CANCELLED =1; /** 表示线程等待唤醒 */ static final int SIGNAL= -1; /** 表示线程等待获取同步锁 */ static final int CONDITION = -2; /** 表示共享模式下无条件传播 */ static final int PROPAGATE = -3;

CANCELLED(1)：表示当前结点已取消调度。当timeout或被中断（响应中断的情况下），会触发变更为此状态，进入该状态后的结点将不会再变化。
SIGNAL(-1)：表示后继结点在等待当前结点唤醒。后继结点入队时，会将前继结点的状态更新为SIGNAL。
CONDITION(-2)：表示结点等待在Condition上，当其他线程调用了Condition的signal()方法后，CONDITION状态的结点将从等待队列转移到同步队列中，等待获取同步锁。
PROPAGATE(-3)：共享模式下，前继结点不仅会唤醒其后继结点，同时也可能会唤醒后继的后继结点。
0：新结点入队时的默认状态。
注意，负值表示结点处于有效等待状态，而正值表示结点已被取消。所以源码中很多地方用>0、<0来判断结点的状态是否正常。