[Java知识库]AbstractQueuedSynchronizer（AQS） 源码细致分析 - 加锁竞争资源分析

1、简介

• AQS 的全称是 AbstractQueuedSynchronizer，它的定位是为 JAVA 种几乎所有的锁和同步器提供一个基础框架
• AQS 是基于 FIFO 队列实现的，并且内部维护了一个状态变量 state，通过原子更新这个状态变量 state 即可实现加锁解锁操作。

主要内部类 Node

static final class Node {
    /** Marker to indicate a node is waiting in shared mode */
    // 枚举：共享模式
    static final Node SHARED = new Node();
    /** Marker to indicate a node is waiting in exclusive mode */
    // 枚举：独占模式
    static final Node EXCLUSIVE = null;

    /** waitStatus value to indicate thread has cancelled */
    // 表示当前节点处于取消状态
    static final int CANCELLED =  1;
    /** waitStatus value to indicate successor's thread needs unparking */
    // 注释：表示当前节点需要唤醒它的后继节点线程 SIGNAL 其实表示的是后继节点的状态，当前节点需要去喊它
    static final int SIGNAL    = -1;
    /** waitStatus value to indicate thread is waiting on condition */
    // 先不说，学习 ReentrantLock 用不到
    static final int CONDITION = -2;
    /**
     * waitStatus value to indicate the next acquireShared should
     * unconditionally propagate
     */
    // 先不说，在ReentrantLock下也用不到
    static final int PROPAGATE = -3;

    // node状态：可选值（0、SIGNAL(-1)、CANCELED(1)、CONDITION、PROPAGATE）
    // waitStatus == 0 默认状态
    // waitStatus > 0：取消状态（在ReentrantLock模式下）
    // waitStatus == -1：表示当前node 如果是 head节点的时候，释放锁之后需要唤醒后继节点
    volatile int waitStatus;

    // 因为node需要构建FIFO 队列，所以 prev就指向当前节点的前驱结点
    volatile Node prev;

    // 因为node需要构建FIFO 队列，所以 prev就指向当前节点的后继结点
    volatile Node next;

    // 当前 node 封装的本身线程
    volatile Thread thread;

    // ReentrantLock 未用到，之后再说（下一个等待在条件上的节点，Condition锁的时候使用）
    Node nextWaiter;

    // 判断当前节点是否是共享模式
    final boolean isShared() {
        return nextWaiter == SHARED;
    }

    // 获取当前节点的前驱节点
    final Node predecessor() throws NullPointerException {
        Node p = prev;
        if (p == null)
            throw new NullPointerException();
        else
            return p;
    }

    // 空构造方法
    Node() {    // Used to establish initial head or SHARED marker
    }

    // 有参构造方法
    Node(Thread thread, Node mode) {     // Used by addWaiter
        // 把共享模式还是互斥模式存储到 nextWaiter 这个字段里面了
        this.nextWaiter = mode;
        this.thread = thread;
    }

    // 有参构造方法
    Node(Thread thread, int waitStatus) { // Used by Condition
        // 等待的状态，在 Condition中使用
        this.waitStatus = waitStatus;
        this.thread = thread;
    }
}

双向链表结构，节点中保存着当前线程、前一个节点、后一个节点以及线程的状态等信息。

主要属性

// AQS 抽象同步队列的成员属性
// 头节点，任何时刻，头结点对应的线程都是当前持锁线程
private transient volatile Node head;

// 阻塞队列的尾部节点 （阻塞队列不包含头结点 head.next -> tail 认为是阻塞队列）
private transient volatile Node tail;

注意：这几个变量都要使用 volatile 关键字来修饰，因为是在多线程环境下操作的，要保证它们的值修改后其他线程立即可见。

这几个变量的修改都是直接使用 Unsafe 这个类来操作的：

// 获取Unsafe类的实例，注意这种方式仅限于jdk自己使用，普通用户是无法这样调用的
private static final Unsafe unsafe = Unsafe.getUnsafe();
// 状态变量state的偏移量
private static final long stateOffset;
// 头节点的偏移量
private static final long headOffset;
// 尾节点的偏移量
private static final long tailOffset;
// 等待状态的偏移量（Node的属性）
private static final long waitStatusOffset;
// 下一个节点的偏移量（Node的属性）
private static final long nextOffset;

static {
    try {
        // 获取state的偏移量
        stateOffset = unsafe.objectFieldOffset
            (AbstractQueuedSynchronizer.class.getDeclaredField("state"));
        // 获取head的偏移量
        headOffset = unsafe.objectFieldOffset
            (AbstractQueuedSynchronizer.class.getDeclaredField("head"));
        // 获取tail的偏移量
        tailOffset = unsafe.objectFieldOffset
            (AbstractQueuedSynchronizer.class.getDeclaredField("tail"));
        // 获取waitStatus的偏移量
        waitStatusOffset = unsafe.objectFieldOffset
            (Node.class.getDeclaredField("waitStatus"));
        // 获取next的偏移量
        nextOffset = unsafe.objectFieldOffset
            (Node.class.getDeclaredField("next"));

    } catch (Exception ex) { throw new Error(ex); }
}

// 调用Unsafe的方法原子更新state
protected final boolean compareAndSetState(int expect, int update) {
    return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

父类属性

AQS 还用到了其父类的 AbstractOwnableSynchronizer 的一些属性：

// 独占模式下：表示当前持有锁的线程~
private transient Thread exclusiveOwnerThread;

2、AQS 成员方法解析

2.1、lock() 方法加锁解析

// 位于 ReentrantLock 静态内部类 Sync 中
abstract static class Sync extends AbstractQueuedSynchronizer {
	abstract void lock();
    ...
}

static final class FairSync extends Sync {  
    // 公平锁入口
    // 不响应中断的加锁 .. 响应中断的加锁方法 lockinterrupted
    final void lock() {
        acquire(1);
    }
    ...
}

2.2、acquire(long arg) 令当前线程竞争资源的方法

// AQS 的accquire方法
public final void acquire(long arg) {
    // 条件一：!tryAcquire(arg) 尝试获取锁，获取成功返回true，获取失败返回false
    // 条件二：2.1：首先是 addWaiter(AbstractQueuedLongSynchronizer.Node.EXCLUSIVE) 将当前线程封装成node然后进行入队操作
    //       2.2：acquireQueued()：挂起当前线程，唤醒后相关的逻辑
    // acquireQueued方法返回true表示挂起过程中线程被中断唤醒过 false：表示未被中断过
    if (!tryAcquire(arg) &&
        // Node.EXCLUSIVE 表示当前节点是独占模式
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        // 再次设置中断标记为 true
        selfInterrupt();
}

2.3、tryAcquire(int acquires) 尝试获取锁的方法

// 位于 ReentrantLock 类的静态内部类 FairSync中：属于尝试获取锁的方法，不会阻塞线程
// 返回 true表示尝试获取锁成功 | 返回 false 表示尝试获取锁失败
// 抢占成功：返回 true 包含重入
// 抢占失败：返回 false
protected final boolean tryAcquire(int acquires) {
    //current：当前线程
    final Thread current = Thread.currentThread();
    // AQS state（是否处于加锁状态） 值
    int c = getState();
    // 条件成立：表示当前 AQS 处于无锁状态
    if (c == 0) {
        // 条件一：
        // 因为 fairSync 是公平锁，任何时候都需要检查一下队列中是否在当前线程之前有等待者...
        // hasQueuedPredecessors() 方法返回 true 表示当前线程前面有等待者
        // hasQueuedPredecessors() 方法返回 false 表示当前线程前面没有等待者，直接尝试获取锁就ok了

        // 条件二：compareAndSetState(0, acquires)
        //  成功：说明当前线程抢占锁成功
        //  失败：说明存在竞争且当前线程竞争失败
        if (!hasQueuedPredecessors() &&
            compareAndSetState(0, acquires)) {
            // 成功之后需要做什么？
            // 设置当前线程为独占者线程
            setExclusiveOwnerThread(current);
            return true;
        }
    }

    // 执行到这里，有几种情况？
    // 1.c != 0 大于 0 的情况，这种情况就需要检查一下 当前线程是不是独占锁的线程，因为 ReentrantLock 是可以重入的
    // 条件成立：说明当前线程就是独占锁线程
    else if (current == getExclusiveOwnerThread()) {
        // 锁重入的逻辑
        // nextc：更新值
        int nextc = c + acquires;
        // 越界判断，当重入的深度很深的时候，会导致 nextc < 0，int 值达到最大值之后再加1 ... 变为负数..
        if (nextc < 0)
            throw new Error("Maximum lock count exceeded");
        // 更新的操作
        setState(nextc);
        return true;
    }

    // 执行到这里，有几种情况：
    // 1.CAS 失败， c == 0的时候，CAS修改 state 的时候未抢过其他线程...
    // 2.c > 0 且 exclusiveOwnerThread != current（当前线程）
    return false;
}

2.4、hasQueuedPredecessors() 判断当前队列中是否有等待者线程

/**
 * true：表示当前线程在队列中有等待者线程
 * flase：表示当前线程前面没有其他等待者线程
 *
 * 调用链：lock -> acquire -> tryAcquire -> (state == 0) hasQueuePredecessor
 *
 * 什么时候返回false呢？
 * 1.当前队列是空
 * 2.当前线程是head.next节点线程  head.next节点线程在任何时候都有权力去争取lock
*/
private boolean hasQueuePredecessor(){
 Node h = head;
 Node t = tail;
 Node s;

 // 条件一：h != t 成立：说明当前队列已经有元素了
 //  不成立：1. h == t == null
 //         2.h == t == head 第一个获取锁失败的线程会为当前持有锁的线程补充创建一个head节点

 // ((s = h.next) == null || s.thread != Thread.currentThread()) 前置条件：条件一成立
 // 排除几种情况：
 // 条件2.1：(s = h.next) == null
 //  极端情况：第一个获取锁失败的线程会为持锁的线程补充创建head头结点，然后再自旋入队 1.casTail 2.pred(node).next  = node
 //  其实想表达的就是：已经有head.next节点了，其他线程在来这的时候需要返回true
 // 条件2.2：s.thread != Thread.currentThread() 前置条件：head.next 不为空
 //  条件成立：说明当前线程就不是head.next节点对应的线程，返回true
 //  条件不成立：说明当前线程就是head.next 节点对应的线程，需要返回false，回头线程会去获取锁了
 return h != t && ((s = h.next) == null || s.thread != Thread.currentThread());
}

2.5、addWaiter(Node mode) 方法：将当前线程加入到阻塞队列中

// AQS 中的 addWaiter() 方法 mode：EXCLUSIVE node 的枚举类型
// 最终会返回当前 线程 包装的node节点
private Node addWaiter(Node mode) {
    // Node.EXCLUSIVE
    // 构建 node，把当前线程封装到对象 node中了
    Node node = new Node(Thread.currentThread(), mode);
    // Try the fast path of enq; backup to full enq on failure
    // 快速入队的过程

    // 获取队尾节点，保存到 pred 变量中
    Node pred = tail;
    // 条件成立：队列中已经有node了
    if (pred != null) {
        // 当前节点的 prev 指向 pred
        node.prev = pred;
        // CAS 成功：说明node入队成功
        if (compareAndSetTail(pred, node)) {
            // 前置节点 next 指向 当前node节点，完成双向绑定
            pred.next = node;
            return node;
        }
    }

    // 什么情况会执行到这里？
    // 1.当前队列是空队列 tail == null
    // 2.CAS 竞争入队 tail 失败，会来到这里

    // 完整入队的逻辑
    enq(node);
    return node;
}

2.6、enq(final Node node) 方法：当前线程对应的节点完整入队的方法（自旋入队）

// AQS 中的 enq()方法
// 完整入队的逻辑
private Node enq(final Node node) {
    // 自旋入队，只有当前node线程入队成功之后才会跳出自旋操作
    for (;;) {
        Node t = tail;
        // 1.当前队列是空队列 tail == null
        // 说明当前锁被占用且当前线程有可能是第一个获取锁失败的线程（当前时刻可能存在一批获取锁失败的线程）
        if (t == null) { // Must initialize
            // 作为当前持锁线程的第一个后继线程，需要做什么事？
            // 1.因为当前持锁的线程，它获取锁的时候，直接 tryAcquire() 成功了，并没有向阻塞队列中添加任何node，所以作为第一个获取锁失败的线程，咱们需要给它擦屁股
            // 2.为自己追加 node节点

            // CAS 成功：说明当前线程成为 head.next节点
            // 当前线程需要为当前持锁的线程创建 node为 head
            if (compareAndSetHead(new Node()))
                tail = head;
            // 注意：这里没有 return，会继续 for自旋操作
        } else {
            // 普通入队方式，只不过在 for中，会保证一定入队成功
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

2.7、acquireQueued(final Node node, long arg) 方法：竞争资源失败后需要做什么？以及挂起被唤醒后需要做的事情（自旋竞争锁资源）

accquireQueued 需要做什么呢？

• 当前节点如果没有被 park 挂起，则 ===> 挂起当前线程
• 线程唤醒后 ===> 需要做一些线程唤醒之后的逻辑

// acquireQueued 需要做什么呢？
// 1.当前节点入队以后有没有被park？挂起？ 没有 => 挂起的操作
// 2.唤醒之后的逻辑在哪呢？ ==> 唤醒之后的逻辑
// AQS 中的 acquireQueued 方法
// 参数一：node 就是当前线程包装出来的 node，且当前时刻已经入队成功了
// 参数二：当前线程抢占资源成功后，设置 state 值，会用到
final boolean acquireQueued(final Node node, long arg) {
    // false：表示当前线程抢占锁成功，普通情况下，lock 当前线程早晚会拿到锁
    // true：表示失败，需要执行出队逻辑（回头讲响应中断的lock逻辑的时候再讲）
    boolean failed = true;
    try {
        // 当前线程是否被中断
        boolean interrupted = false;
        // 自旋操作
        for (;;) {
            // 什么时候会执行这里？
            // 1.进入 for循环的时候，在线程尚未 park前会执行
            // 2.唤醒线程 park 之后被唤醒的时候，会进入到这里

            // predecessor()：获取当前线程node节点的前置节点
            final Node p = node.predecessor();

            // 条件一成立：说明当前节点为head.next 节点 head.next 节点在任何时候都有权力去争夺锁
            // 条件二：tryAcquire(arg)、
            // 成立：说明上一个head 对应的线程已经释放锁了，head.next 节点对应的线程正好获取到锁了
            // 不成立：说明 head 对应的线程还未释放锁呢 head.next 仍然需要被park...
            if (p == head && tryAcquire(arg)) {
                // 拿到锁之后需要做什么？
                // 设置当前节点为head节点
                setHead(node);
                // 将上个线程对应的node.next引用置为空，协助老的head出队
                p.next = null; // help GC
                // 当前线程获取锁过程中没有发生异常
                failed = false;
                // 返回当前线程的中断标记
                return interrupted;
            }

            // shouldParkAfterFailedAcquire(p, node)：这个方法是干什么的呢？ 当前线程获取锁自旋失败后是否需要挂起呢?
            //  返回值 true：表示当前线程需要挂起 false：不需要挂起
            // parkAndCheckInterrupt()：前提条件：当前线程需要挂起 这个方法什么作用？
            //  挂起当前线程，并且唤醒之后返回当前线程的中断标记
            //   (唤醒：1.正常唤醒 其他线程 unpark 2.其他线程给当前挂起的线程一个中断信号也会被唤醒)
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                // 进入到这里面，说明当前线程是被其他线程的中断信号给唤醒的
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

2.8、shouldParkAfterFailedAcquire(Node pred, Node node) 方法：判断当前线程资源竞争失败后需不需要被挂起？

/**
 * 作用：当前线程获取锁自旋失败后是否需要挂起呢?
 * @param pred 线程 node 的前置节点
 * @param node 当前线程对应的 node
 * @return true：表示当前线程需要挂起 false：表示当前线程不需要挂起
 *
 * 总结：
 * 1.当前节点的前置节点是 取消状态 第一次来到这个方法的时候，会越过取消状态的节点，第二次或者第三次会返回 true，然后park当前线程
 * 2.当前节点的前置结点状态是0，当前线程会设置前置节点的状态为-1，表示 SIGNAL状态，唤醒它的第一个后继节点，第二次自旋来到这个方法的时候，会返回true，park当前节点
 */
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    // 获取前置节点的状态
    // waitStatus：0：默认状态 new Node() -1：SIGNAL 表示当前节点释放锁之后会唤醒它的第一个后继节点
    // > 0：表示当前节点是 CANCELED 状态 取消状态
    int ws = pred.waitStatus;
    // 条件成立：表示前置节点是个可以唤醒当前节点的节点，所以返回 true
    // 普通情况下：第一个来到shouldParkAfterFailedAcquire ws 不会是 -1
    if (ws == Node.SIGNAL) // waitStatus == -1
        return true;

    // 条件成立：表示前置节点是 CANCELED 节点
    if (ws > 0) {
        // 找爸爸的过程：条件是什么呢？ 前置节点的 waitStatus <= 0的情况
        // 依次找到队列中当前节点node的所有前驱节点中 waitStatus <= 0 的节点
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        // 再前驱节点中找到了 waitStatus <= 0 的节点
        // 隐含着一种操作，CANCELED 状态的节点会被出队
        pred.next = node;
    } else { // waitStatus == 0
        // 进入到这里，说明当前节点的前驱节点的状态 waitStatus == 0的状态
        // 将当前node的前置节点node waitStatus 状态强制设置为 SIGNAL，表示前置节点释放锁之后需要唤醒当前节点
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

总结：

• 如果当前节点的前置节点是 CANCLED 取消状态，则：
  • 第一次自旋来到这个方法的时候，会越过取消状态的节点。
  • 第二次返回 true，然后 park 当前线程
• 如果当前节点的前置节点是 0 默认状态，则：
  • 当前线程节点会设置前置节点的状态为 -1（SIGNAL）
  • 第二次自旋来到这个方法的时候，会返回 true，然后 park 挂起当前线程

2.9、parkAndCheckInterrupt() 挂起当前线程

// AQS 中的 方法
// 这个方法作用是 park当前线程，唤醒后返回当前线程是否为中断信号唤醒
private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
}