目录

前言

定义

AQS 的核心

volatile int state

CLH 队列的变体

状态与同步队列的关系

CLH 队列变体

入队操作

acquire方法

addWaiter方法

enq方法

acquireQueued方法

shouldParkAfterFailedAcquire方法

parkAndCheckInterrupt方法

需要注意的地方

出队操作

release方法

为什么检查头节点的waitStatus？

unparkSuccessor方法

为什么从后向前遍历？

---

前言

上一篇文章我们了解了显示锁ReentrantLock的使用（传送门：https://blog.csdn.net/weixin_38357164/article/details/157143068?fromshare=blogdetail&sharetype=blogdetail&sharerId=157143068&sharerefer=PC&sharesource=weixin_38357164&sharefrom=from_link）今天我们要讲的是AQS队列含义，以及AQS队列源码的部分解析，希望通过阅读本文可以帮助你更好的理解java锁的实现原理，更好的在实际开发中运用java显示锁，本文为AQS队列解析系列第一章（由于篇幅有限不含全部源码解读），后续会逐步更新，敬请期待！

定义

AQS（AbstractQueuedSynchronizer）是Java并发包的核心框架，用于构建锁和其他同步器的基础框架。Java 并发包（java.util.concurrent）中许多同步器（如 ReentrantLock, Semaphore, CountDownLatch, ReentrantReadWriteLock 等）所使用的核心基础框架。

AQS的核心思想是：如果被请求的共享资源空闲，则将当前请求资源的线程设置为有效的工作线程，并将共享资源设置为锁定状态；如果共享资源被占用，就需要一套线程获取锁、阻塞等待、以及被唤醒继续获取锁的队列机制来控制线程的同步，这个机制就是AQS 队列。

AQS 的核心

volatile int state

一个同步状态变量。使用 volatile 关键字确保多线程环境下对 state 的修改可见性，即一个线程修改了 state，其他线程能立即看到最新值。它的含义由子类定义。

• 当 ReentrantLock 为例，它是基于 AQS 实现的独占锁。在 ReentrantLock 中，state 表示锁的重入次数：

  • 当 state = 0 时，表示锁没有被任何线程持有。

  • 当 state > 0 时，表示锁被某个线程持有，并且该线程重入了 state 次。

• 在 Semaphore 中，state 表示剩余的许可证数量。
• 在 CountDownLatch 中，state 表示计数器的值。
• ......

CLH 队列的变体

一个 （FIFO 双向队列）：用于管理等待线程。这个队列是 AQS 的灵魂，我们接下来重点讲解。

状态与同步队列的关系

• 当线程尝试获取同步状态时，会先读取 state 的值，根据同步器的规则判断是否能够获取。

  • 如果获取成功（通常通过 CAS 设置 state），则线程继续执行。

  • 如果获取失败，则将该线程封装成节点（Node）并加入同步队列。

• 当释放同步状态时，会修改 state 的值，并唤醒队列中等待的线程，使其重新尝试获取状态。

CLH 队列变体

我们本期只关注AQS队列的实现以及源码解读，至于状态与AQS同步队列的关系如何关联，后续会在juc工具类的相关源码解读中为大家呈现。

在AQS队列中每一个线程都绑定为一个node对象。Node 类的核心字段如下

static final class Node {
    // 共享模式下的节点标记
    static final Node SHARED = new Node();
    // 独占模式下的节点标记
    static final Node EXCLUSIVE = null;

    // 等待状态，初始为0
    volatile int waitStatus;

    // 前驱节点
    volatile Node prev;

    // 后继节点
    volatile Node next;

    // 等待在这个节点上的线程
    volatile Thread thread;

    // 用于链接到下一个等待节点（可能是共享模式或独占模式）
    Node nextWaiter;

    // ... 其他方法和常量
}

• waitStatus：这是节点状态的关键。
  • SIGNAL (-1)：表示当前节点的后继节点需要被唤醒（通常在当前节点释放锁时设置）。
  • CANCELLED (1)：表示节点因为超时或中断而被取消，需要从队列中移除。
  • CONDITION (-2)：用于条件队列（Condition）。
  • PROPAGATE (-3)：用于共享模式下的传播。
  • 0：初始状态。
• prev 和 next：构成了双向链表，用于实现出队和入队操作。
• nextWaiter：在独占模式下为 null，在共享模式下指向下一个共享节点（用于实现共享模式的唤醒传播）。

入队操作

acquire方法

本文只解析acquire(int arg)方法，该方法不支持中断，不支持超时。由于篇幅原因和个人精力有限，后续AQS解析篇章中会逐步解析到，还请谅解。

    public final void acquire(int arg) {
        //tryAcquire需由实现重写
        if (!tryAcquire(arg) &&
            //如果tryAcquire没有获取锁，则线程加入队列并决定是否挂起线程，后续唤醒后继续争抢锁。
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            //如果执行到这里，那么线程会在队列里清除过中断状态，这里需重新设置一下线程的中断状态。
            selfInterrupt();
    }

先执行 tryAcquire(arg) 尝试获取锁，如果失败再执行acquireQueued(addWaiter(Node.EXCLUSIVE), arg))方法，首先看addWaiter(Node.EXCLUSIVE), arg)方法。

addWaiter方法

private Node addWaiter(Node mode) {// mode为独占锁模式
    // 封装当前线程为node对象，模式为EXCLUSIVE
    Node node = new Node(Thread.currentThread(), mode);
    // 获取此时尾指针的引用pred，假设在读取tail和后续操作之间，队列尾部没有变化（而实际可能有变化）
    Node pred = tail;
    if (pred != null) {
        //node的前驱指向tail，这一步不是原子操作但没有问题，因为即使并发修改后续会通过enq方法纠正。
        node.prev = pred;
        //compareAndSetTail保证了操作尾指针的原子性，如果预期的pred和此刻Tail相同时执行，防止多线程下tail被更新
        // 尝试一次node入队，如CAS乐观锁成功就直接返回，不成功就使用enq(node)自旋设置到队列尾
        if  (compareAndSetTail(pred, node)) {
            //tail的next指向当前节点node
            pred.next = node;
            return node;
        }
    }
    //如果tail==null或cas失败，就使用enq(node)自旋设置到队列尾
    enq(node);
    return node;
}

enq方法

private Node enq(final Node node) {
    //自旋
    for (;;) {
        //取当前tail，并发下tail可能会被修改。
        Node t = tail;
        //如果tail==null创建一个新的节点并添加到队列头节点
        if (t == null) {
            if (compareAndSetHead(new Node()))
                //该node的head和tail都指向这个虚拟头节点（因为这个虚拟节点没有绑定线程）
                tail = head;
        } else {
            //tail != null，将tail和node双向绑定
            //先绑定prev，node前驱指向tail
            node.prev = t;
            //CAS成功再绑定next，如果不成功则线程自旋转直到成功，成功后一定会在队列尾部
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

acquireQueued方法

执行完addWaiter方法后，再执行acquireQueued方法。线程加入队列之后，就要尝试获取锁了，如果没有获取就挂起线程。

参数node：当前加入队列尾的node。

参数arg：加锁次数1。

final boolean acquireQueued(final Node node, int arg) {
    //自旋过程中是否异常退出
    boolean failed = true;
    try {
        //判断自旋过程中是否被中断过
        boolean interrupted = false;
        for (;;) {
            //获取node前驱节点 P
            final Node p = node.predecessor();
            //如果P是头节点，node就尝试获取一次锁，如果不成功就自旋获取
            if (p == head && tryAcquire(arg)) {
                //成功说明head已经释放锁，将当前node设置为head
                setHead(node);
                //原head与链表断开链接，help gc
                p.next = null；
                failed = false；
                return interrupted;
            }
            // p不是head
            // 或p是head但tryAcquire(arg)失败（失败的原因是原head释放锁，非公平锁情况下会有链表外的线程直接参与竞争锁并成功）
            // 根据前驱节点的状态，执行shouldParkAfterFailedAcquire判断是否应该阻塞当前线程
            // 如果需要阻塞则继续执行parkAndCheckInterrupt。
            if (shouldParkAfterFailedAcquire(p, node) &&  parkAndCheckInterrupt())
                // 如果线程node应该被park，且设置过中断状态，此时interrupted=true;
                interrupted = true;
        }
    } finally {
        //如果异常的话则取消该节点node的尝试获取锁的操作
        if (failed)
            cancelAcquire(node);
    }
}

shouldParkAfterFailedAcquire方法

private static boolean shouldParkAfterFailedAcquire(AbstractQueuedSynchronizer.Node pred, AbstractQueuedSynchronizer.Node node) {
    //获取node前驱节点pred的状态waitStatus
    int ws = pred.waitStatus;
    // 前驱节点状态为SIGNAL，表示前驱节点释放锁后会唤醒node
    if (ws == AbstractQueuedSynchronizer.Node.SIGNAL)
        //当前node节点可以安全阻塞
        return true;
    //如前驱节点被取消（CANCELLED）
    if (ws > 0) {
        //跳过队列中所有被取消的前驱节点（清理队列）
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        //清理后重新绑定前驱节点和node
        pred.next = node;
    //其他情况（0或-2），尝试将前驱节点状态设置为 SIGNAL，下次循环可能就会阻塞
    } else {
        // 前驱节点状态为0或CONDITION，尝试设置为SIGNAL
        compareAndSetWaitStatus(pred, ws, AbstractQueuedSynchronizer.Node.SIGNAL);
    }
    // 不阻塞，继续循环
    return false;
}

parkAndCheckInterrupt方法

private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);  // 阻塞当前线程
    return Thread.interrupted();  // 检查是否被中断，并清除中断状态
}

需要注意的地方

第一：休眠当前线程，注意当 unpark 的时候线程也从此继续执行！

第二：park方法阻塞线程的必要条件：未处于中断状态、无permit，如果线程中断标志位为true，而在该状态下LockSurport.park方法并不会生效，使得程序继续执行。若该线程始终获取不到锁，该线程将在acquireQueue方法的循环中空转，cpu有可能会出现100%，所以如果park不成功就要执行Thread.interrupted()清除当前线程的中断状态让其可以park。

出队操作

release方法

public final boolean release(int arg) {
    //当前线程即将释放锁
    //tryRelease是一个由子类实现的方法
        true：表示锁被完全释放（例如在ReentrantLock中，state变为0）
        false：表示锁未被完全释放（例如可重入锁中，锁仍被当前线程持有）
    if (tryRelease(arg)) {
        //成功释放锁后取队列头结点（队列中持有锁的节点一定是头节点）
        Node h = head;
        //头结点不为null并且等待状态不为0
        if (h != null && h.waitStatus != 0)
            // 唤醒头节点的下一个节点
            unparkSuccessor(h);
        return true;
    }
    return false;
}

为什么检查头节点的waitStatus？

• 头节点的 waitStatus 通常表示其后继节点的状态
• 当头节点的后继节点需要被唤醒时，头节点的 waitStatus 会被设置为 Node.SIGNAL (-1)
• 如果 waitStatus == 0，说明没有后继节点需要被唤醒

unparkSuccessor方法

private void unparkSuccessor(Node node) {
    //头节点的ws状态
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);  // 将节点状态重置为0，node.waitStatus 通常是 SIGNAL(-1)，表示该节点有责任唤醒后继。重置为0表示"我已履行唤醒后继节点的使命" 
    //获取后继节点s   
    Node s = node.next;  
    // 如果头节点的第一个后继节点为null或已取消
    if (s == null || s.waitStatus > 0) {  
        
        s = null;
        // 从尾节点向前遍历，找到最前面的未取消节点，赋值给s
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
    //唤醒s线程
    if (s != null)
        LockSupport.unpark(s.thread); 
}

为什么从后向前遍历？

因为会考虑入队时候的安全问题，线程节点在入队的时候，是先执行的node.prev = tail。然后如果CAS成功后再执行，tail.next=node。现在假设有下面这中情况。

1. 线程A：执行 node.prev = tail（步骤1）

2. 线程A：执行 compareAndSetTail(t, node) 成功（步骤2）

3. 此时线程B调用 unparkSuccessor：

   • tail.next 还未设置（步骤3未执行）

   • 此时直接通过 tail.next 指针找不到新节点(会是null)

4. 线程A：执行 tail.next = node（步骤3，可能在其他线程唤醒操作unparkSuccessor之后）

 ======================================

喜欢请点赞收藏加关注~~~

下一篇我们继续讲AQS队列第二章，敬请期待！！！

26年2月1日更新：https://blog.csdn.net/weixin_38357164/article/details/157582601?fromshare=blogdetail&sharetype=blogdetail&sharerId=157582601&sharerefer=PC&sharesource=weixin_38357164&sharefrom=from_link

=======================================