当前位置: 首页 > web >正文

深入解析AQS:Java并发核心框架

我们来深入解析一下 java.util.concurrent.locks.AbstractQueuedSynchronizer (简称 AQS)。

AbstractQueuedSynchronizer 是 Java 并发包中非常核心的一个基础框架,它为实现依赖于先进先出 (FIFO) 等待队列的阻塞锁和相关的同步器(如信号量、事件等)提供了一个标准化的机制。许多我们熟知的并发工具类,例如 ReentrantLockSemaphoreCountDownLatchReentrantReadWriteLock 等,都是基于 AQS 实现的。

AQS 的核心设计思想

AQS 的核心思想是

  • 如果被请求的共享资源空闲,则将当前请求资源的线程设置为有效的工作线程,并将共享资源设置为锁定状态。
  • 如果共享资源被占用,就需要一定的阻塞等待唤醒机制来保证锁分配的公平性,并将暂时获取不到锁的线程加入到队列中。

AQS 使用一个整型的 volatile 变量(state)来表示同步状态,并通过内置的 FIFO 队列来完成线程的排队工作。子类通过继承 AQS 并重写指定的方法来管理这个状态,从而定义锁的获取和释放逻辑。

AQS 的结构

AQS 的主要结构包含以下几个部分:

  1. state (同步状态):

    • 这是一个 volatile int 类型的变量,用于表示当前的同步状态。
    • 它的具体含义由子类定义。例如,在 ReentrantLock 中,state 表示锁的重入次数;在 Semaphore 中,state 表示可用的许可数量。
    • AQS 提供了 getState()setState(int newState) 和 compareAndSetState(int expect, int update) 这三个受保护的方法来原子地读取和修改 state

    AbstractQueuedSynchronizer.java

    // ... existing code ...
    private volatile int state;/*** Returns the current value of synchronization state.* This operation has memory semantics of a {@code volatile} read.* @return current state value*/
    protected final int getState() {return state;
    }/*** Sets the value of synchronization state.* This operation has memory semantics of a {@code volatile} write.* @param newState the new state value*/
    protected final void setState(int newState) {state = newState;
    }/*** Atomically sets synchronization state to the given updated* value if the current state value equals the expected value.* This operation has memory semantics of a {@code volatile} read* and write.** @param expect the expected value* @param update the new value* @return {@code true} if successful. False return indicates that the actual*         value was not equal to the expected value.*/
    protected final boolean compareAndSetState(int expect, int update) {return U.compareAndSetInt(this, STATE, expect, update);
    }
    // ... existing code ...
    
  2. 等待队列 (Wait Queue):

    • AQS 内部维护一个 FIFO 的双向链表队列,用于存放等待获取同步状态的线程。
    • 这个队列是 CLH (Craig, Landin, and Hagersten) 锁队列的一个变体。CLH 锁通常用于自旋锁【CLH锁使得每个节点的不对同一内存进行CAS操作,减少了CAS风暴(缓存行在不同处理器核心间频繁失效和同步))】,但 AQS 将其用于阻塞同步器。
    • 队列的头节点 (head) 是一个虚拟节点(dummy node),它不代表任何等待的线程,主要用于简化队列操作。真正的第一个等待线程是 head.next
    • 尾节点 (tail) 指向队列中最后一个等待的线程节点。
    • 线程入队和出队的操作是原子性的,主要通过 CAS (Compare-And-Swap) 操作 tail 和 head 指针实现。

    AbstractQueuedSynchronizer.java

    // ... existing code ...
    /*** Head of the wait queue, lazily initialized.*/
    private transient volatile Node head;/*** Tail of the wait queue. After initialization, modified only via casTail.*/
    private transient volatile Node tail;
    // ... existing code ...
    
  3. Node (队列节点):

    • 队列中的每个节点 (Node) 都封装了一个等待获取同步状态的线程 (waiter)。
    • Node 包含以下重要字段:
      • volatile int status: 节点状态,用于控制线程的阻塞和唤醒。常见的状态有:
        • WAITING (1): 表示线程正在等待被唤醒。后继节点需要唤醒当前节点。
        • CANCELLED (0x80000000): 表示线程因为超时或中断而被取消。
        • COND (2): 表示线程正在条件队列中等待。
        • 0: 初始状态或表示线程已释放。
      • volatile Node prev: 指向前驱节点。
      • volatile Node next: 指向后继节点。
      • Thread waiter: 节点关联的线程。
    • AQS 定义了 Node 的几个子类:ExclusiveNode (独占模式节点) 和 SharedNode (共享模式节点),以及用于条件变量的 ConditionNode

    AbstractQueuedSynchronizer.java

    // ... existing code ...
    /** CLH Nodes */
    abstract static class Node {volatile Node prev;       // initially attached via casTailvolatile Node next;       // visibly nonnull when signallableThread waiter;            // visibly nonnull when enqueuedvolatile int status;      // written by owner, atomic bit ops by others// methods for atomic operationsfinal boolean casPrev(Node c, Node v) {  // for cleanQueuereturn U.weakCompareAndSetReference(this, PREV, c, v);}final boolean casNext(Node c, Node v) {  // for cleanQueuereturn U.weakCompareAndSetReference(this, NEXT, c, v);}final int getAndUnsetStatus(int v) {     // for signallingreturn U.getAndBitwiseAndInt(this, STATUS, ~v);}final void setPrevRelaxed(Node p) {      // for off-queue assignmentU.putReference(this, PREV, p);}final void setStatusRelaxed(int s) {     // for off-queue assignmentU.putInt(this, STATUS, s);}final void clearStatus() {               // for reducing unneeded signalsU.putIntOpaque(this, STATUS, 0);}private static final long STATUS= U.objectFieldOffset(Node.class, "status");private static final long NEXT= U.objectFieldOffset(Node.class, "next");private static final long PREV= U.objectFieldOffset(Node.class, "prev");
    }// Concrete classes tagged by type
    // 这两个只是标记类型
    static final class ExclusiveNode extends Node { }
    static final class SharedNode extends Node { }static final class ConditionNode extends Nodeimplements ForkJoinPool.ManagedBlocker {ConditionNode nextWaiter;            // link to next waiting node/*** Allows Conditions to be used in ForkJoinPools without* risking fixed pool exhaustion. This is usable only for* untimed Condition waits, not timed versions.*/public final boolean isReleasable() {return status <= 1 || Thread.currentThread().isInterrupted();}public final boolean block() {while (!isReleasable()) LockSupport.park();return true;}}
    // ... existing code ...
    
  4. ConditionObject (条件变量):

    • AQS 提供了一个内部类 ConditionObject,它是 Condition 接口的实现。
    • ConditionObject 使得线程可以在某个条件不满足时挂起(await),并在条件满足时被其他线程唤醒(signal 或 signalAll)。
    • 每个 ConditionObject 内部也维护一个 FIFO 的等待队列(与 AQS 的主等待队列不同),用于存放因该条件而等待的线程。
    • 条件变量只能在独占模式下使用。

    AbstractQueuedSynchronizer.java

    // ... existing code ...
    static final class ConditionNode extends Nodeimplements ForkJoinPool.ManagedBlocker {ConditionNode nextWaiter;            // link to next waiting node
    // ... existing code ...
    /*** Condition implementation for a {@link* AbstractQueuedSynchronizer} serving as the basis of a {@link* Lock} implementation.** <p>This class provides behavior equivalent to the implicit* monitors accessed using {@code synchronized} methods and* statements, but offers extends capabilities.  In particular,* multiple {@code Condition} objects can be associated with a single* {@code Lock}, and they can be used in conjunction with either* fair or unfair locks.** <p>The lock associated with this condition is the* AbstractQueuedSynchronizer instance that it is constructed for.* When an await method is called, the lock must be held by the* current thread.  A {@link IllegalMonitorStateException} will be* thrown if this is not the case.** <p>When the await methods return, the lock is always held by the* current thread.** <p>This class is Serializable, but requires that the underlying AQS* lock is also serializable.** @since 1.5*/
    public class ConditionObject implements Condition, java.io.Serializable {
    // ... existing code ...
    

AQS 的功能模式

AQS 支持两种同步模式:

  1. 独占模式 (Exclusive Mode):

    • 同一时刻只允许一个线程获取同步状态。
    • 例如 ReentrantLock
    • 子类需要重写 tryAcquire(int arg) 和 tryRelease(int arg) 方法。
    • 如果需要支持条件变量,还需要重写 isHeldExclusively() 方法。
  2. 共享模式 (Shared Mode):

    • 同一时刻允许多个线程获取同步状态。
    • 例如 SemaphoreCountDownLatchReentrantReadWriteLock 的读锁。
    • 子类需要重写 tryAcquireShared(int arg) 和 tryReleaseShared(int arg) 方法。

一个同步器可以只支持其中一种模式,也可以同时支持两种模式(例如 ReentrantReadWriteLock)。

AQS 的核心方法 (需要子类实现)

子类通过重写以下受保护的方法来定义其同步语义:

  • protected boolean tryAcquire(int arg): 尝试以独占模式获取资源。如果成功则返回 true,否则返回 false。此方法应检查状态,并在成功获取时更新状态。
  • protected boolean tryRelease(int arg): 尝试以独占模式释放资源。如果成功则返回 true,否则返回 false。此方法应检查状态,并在成功释放时更新状态。
  • protected int tryAcquireShared(int arg): 尝试以共享模式获取资源。返回一个负值表示失败;0 表示成功但后续共享模式获取将失败;正值表示成功且后续共享模式获取可能也会成功。
  • protected boolean tryReleaseShared(int arg): 尝试以共享模式释放资源。如果成功则返回 true,否则返回 false
  • protected boolean isHeldExclusively(): 判断当前线程是否独占资源。

这些方法的实现必须是线程安全的,并且通常应该简短且不阻塞。

AQS 的工作流程 (以独占模式为例)

获取同步状态 (Acquire)

当一个线程调用 AQS 子类提供的 acquire(int arg) 方法(或其他获取方法如 acquireInterruptiblytryAcquireNanos)时,大致流程如下:

  1. 调用 tryAcquire(arg): 尝试获取锁。
    • 如果 tryAcquire 返回 true (获取成功),则方法直接返回。
    • 如果 tryAcquire 返回 false (获取失败),则进入下一步。
  2. 创建节点并入队:
    • 如果当前线程的节点尚未创建,则创建一个新的 Node (通常是 ExclusiveNode),并将当前线程 (Thread.currentThread()) 设置为其 waiter
    • 将该节点通过 CAS 操作加入到等待队列的尾部。
    • 在入队过程中,如果队列未初始化(head 为 null),会先调用 tryInitializeHead() 初始化头节点。
  3. 阻塞当前线程:
    • 入队成功后,当前线程会进入一个循环。
    • 在循环中,线程会检查自己是否是队列中的第一个有效等待者(即前驱节点是 head)。
    • 如果是第一个等待者,会再次尝试调用 tryAcquire(arg)
    • 如果不是第一个等待者,或者再次尝试 tryAcquire 失败,则会检查前驱节点的状态。
      • 如果前驱节点状态为 WAITING,则当前线程可以通过 LockSupport.park(this) 安全地阻塞自己,等待被前驱节点唤醒。
      • 如果前驱节点状态为 CANCELLED,则向前遍历并移除已取消的节点。
    • 在阻塞前,通常会将当前节点的状态设置为 WAITING,表示它需要被唤醒。
  4. 唤醒与重试:
    • 当持有锁的线程释放锁时(调用 release),它会唤醒队列中头节点的后继节点(如果存在且状态正常)。
    • 被唤醒的线程从 LockSupport.park() 返回,清除自身的中断状态(如果被中断),然后回到步骤 3 的循环,再次尝试获取锁。
    • 如果获取成功,它会将自己设置为新的 head 节点,并将旧的 head 节点出队(通过将其 next 指针置为 null)。

AQS 的 acquire 方法 (final int acquire(Node node, int arg, boolean shared, boolean interruptible, boolean timed, long time)) 是所有获取操作的核心实现,它处理了上述复杂的逻辑,包括节点的创建、入队、自旋尝试、阻塞、中断处理、超时处理以及 OutOfMemoryError 的处理。

AbstractQueuedSynchronizer.java

// ... existing code ...final int acquire(Node node, int arg, boolean shared,boolean interruptible, boolean timed, long time) {Thread current = Thread.currentThread();byte spins = 0, postSpins = 0;   // retries upon unpark of first threadboolean interrupted = false, first = false;Node pred = null;               // predecessor of node when enqueued/** Repeatedly:*  Check if node now first*    if so, ensure head stable, else ensure valid predecessor*  if node is first or not yet enqueued, try acquiring*  else if queue is not initialized, do so by attaching new header node*     resort to spinwait on OOME trying to create node*  else if node not yet created, create it*     resort to spinwait on OOME trying to create node*  else if not yet enqueued, try once to enqueue*  else if woken from park, retry (up to postSpins times)*  else if WAITING status not set, set and retry*  else park and clear WAITING status, and check cancellation*/for (;;) {if (!first && (pred = (node == null) ? null : node.prev) != null &&!(first = (head == pred))) { // 检查当前节点是否是第一个等待者 (head的后继)if (pred.status < 0) { // 如果前驱节点被取消cleanQueue();           // predecessor cancelledcontinue;} else if (pred.prev == null) { // 前驱节点是head,但head还没稳定Thread.onSpinWait();    // ensure serializationcontinue;}}if (first || pred == null) { // 如果是第一个等待者,或者还没有前驱 (意味着可以尝试获取)boolean acquired;try {if (shared)acquired = (tryAcquireShared(arg) >= 0);elseacquired = tryAcquire(arg); // 尝试获取锁} catch (Throwable ex) {cancelAcquire(node, interrupted, false);throw ex;}if (acquired) { // 获取成功if (first) { // 如果是之前排队的第一个节点获取成功node.prev = null;head = node; // 将当前节点设置为新的headpred.next = null; // 旧的head出队node.waiter = null;if (shared)signalNextIfShared(node); // 如果是共享模式,可能需要唤醒后继if (interrupted)current.interrupt(); // 恢复中断状态}return 1; // 返回1表示获取成功}}Node t;if ((t = tail) == null) {           // 如果队列未初始化if (tryInitializeHead() == null) // 初始化队列 (创建dummy head)return acquireOnOOME(shared, arg); // OOM处理} else if (node == null) {          // 如果节点还未创建 (通常是第一次尝试失败后)try {node = (shared) ? new SharedNode() : new ExclusiveNode(); // 创建节点} catch (OutOfMemoryError oome) {return acquireOnOOME(shared, arg); // OOM处理}} else if (pred == null) {          // 如果节点已创建但还未入队 (pred为null表示之前casTail失败或新创建)node.waiter = current;node.setPrevRelaxed(t);         // 设置前驱为当前tailif (!casTail(t, node))          // CAS尝试将当前节点设置为新的tailnode.setPrevRelaxed(null);  // CAS失败,回滚elset.next = node;              // CAS成功,旧tail的next指向新tail} else if (first && spins != 0) { // 如果是第一个节点被唤醒,进行少量自旋尝试--spins;Thread.onSpinWait();} else if (node.status == 0) { // 如果节点状态为0 (初始状态)node.status = WAITING;          // 设置为WAITING,表示需要被唤醒} else { // 准备阻塞spins = postSpins = (byte)((postSpins << 1) | 1); // 更新自旋计数器 (指数退避)try {long nanos;if (!timed)LockSupport.park(this); // 阻塞当前线程else if ((nanos = time - System.nanoTime()) > 0L)LockSupport.parkNanos(this, nanos); // 带超时的阻塞elsebreak; // 超时} catch (Error | RuntimeException ex) {cancelAcquire(node, interrupted, interruptible); // 发生异常,取消获取并重抛throw ex;}node.clearStatus(); // 被唤醒后清除状态if ((interrupted |= Thread.interrupted()) && interruptible) // 检查中断break; // 如果可中断且已中断,则退出}}return cancelAcquire(node, interrupted, interruptible); // 处理取消或中断
// ... existing code ...

释放同步状态 (Release)

当一个线程调用 AQS 子类提供的 release(int arg) 方法时,大致流程如下:

  1. 调用 tryRelease(arg): 尝试释放锁。
    • 如果 tryRelease 返回 false (释放失败,例如当前线程并非持有者或状态不匹配),则方法直接返回 false (或抛出异常,具体取决于子类实现)。
    • 如果 tryRelease 返回 true (释放成功),则进入下一步。
  2. 唤醒后继节点:
    • 获取当前 head 节点。
    • 如果 head 不为 null 且其状态不为 0 (通常意味着有后继节点在等待),则调用 signalNext(head) (或 signalNextIfShared(head) 如果是共享模式) 来唤醒 head 的后继节点。
    • signalNext 会找到 head 的第一个有效(非 CANCELLED)的后继节点 s,将其状态 status 中的 WAITING 位清除,然后调用 LockSupport.unpark(s.waiter) 唤醒该节点关联的线程。

AbstractQueuedSynchronizer.java

// ... existing code .../*** 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)) {signalNext(head);return true;}return false;}
// ... existing code ...private static void signalNext(Node h) {Node s;if (h != null && (s = h.next) != null && s.status != 0) {s.getAndUnsetStatus(WAITING); // 清除后继节点的WAITING状态LockSupport.unpark(s.waiter); // 唤醒后继节点的线程}}// WAITING==1 取反是-2,但是这个CAS还取了&,因此变为0final int getAndUnsetStatus(int v) {     // for signallingreturn U.getAndBitwiseAndInt(this, STATUS, ~v);}public final int getAndBitwiseAndInt(Object o, long offset, int mask) {int current;do {current = getIntVolatile(o, offset);} while (!weakCompareAndSetInt(o, offset,current, current & mask));return current;}
// ... existing code ...

head

head 节点既可以是“哑节点”(dummy node / sentinel node),也可以是代表“当前持有锁的线程”的节点,但关键在于 head 节点本身从不代表一个正在等待的线程。我们来分解一下 head 的状态和作用:

  1. 队列未初始化时 (head == null):

    • 最初,当没有线程竞争锁,或者AQS实例刚创建时,head 和 tail 都是 null
  2. 首次发生竞争,队列初始化时 (tryInitializeHead):

    • 当第一个线程尝试获取锁失败并需要入队时,或者条件队列中的节点转移到同步队列时,如果队列尚未初始化,会调用 tryInitializeHead()
    • 在这个方法内部,会创建一个新的 Node 对象(通常是 ExclusiveNode 或 SharedNode),并将其设置为 head。此时,tail 也指向这个新创建的 head 节点。

      AbstractQueuedSynchronizer.java

      // ... existing code ...
      private Node tryInitializeHead() {for (Node h = null, t;;) {if ((t = tail) != null)return t;else if (head != null)Thread.onSpinWait();else {if (h == null) {try {h = new ExclusiveNode(); // 创建一个新的节点} catch (OutOfMemoryError oome) {return null;}}if (U.compareAndSetReference(this, HEAD, null, h)) // 将新节点设置为 headreturn tail = h; // tail 也指向这个 head}}
      }
      // ... existing code ...
      
    • 在这个阶段,head 节点是一个“哑节点”或“哨兵节点”。 它不关联任何实际的等待线程,它的 waiter 字段是 nullstatus 通常是 0。它的主要作用是简化队列操作的边界条件,使得队列的第一个实际等待节点始终是 head.next
  3. 当一个等待的线程成功获取锁后:

    • 当队列中的某个节点(比如之前的 head.next)成功获取到锁时,这个节点会成为新的 head
    • 这个过程发生在 acquire 方法的核心逻辑中:

      AbstractQueuedSynchronizer.java

      // ... existing code ...
      final int acquire(Node node, int arg, boolean shared,boolean interruptible, boolean timed, long time) {
      // ... existing code ...if (acquired) { // 如果当前节点成功获取了锁if (first) { // 'first' 为 true 意味着当前节点是队列的第一个有效等待者 (即旧 head 的后继)node.prev = null; // 新 head 的 prev 指向 nullhead = node;      // 当前成功获取锁的节点成为新的 headpred.next = null; // pred 是旧的 head,将其 next 指向 null (帮助 GC,使其出队)node.waiter = null; // 新 head 的 waiter 设置为 null,因为它不再等待if (shared)signalNextIfShared(node);if (interrupted)current.interrupt();}return 1;}
      // ... existing code ...
      }
      
    • 在这个阶段,head 节点代表了“当前持有锁的线程”所对应的节点。 尽管它代表持有锁的线程,但它的 waiter 字段会被设为 null,因为这个线程已经不再处于等待状态。它的 status 通常也是 0

总结与关键点:

  • head 的“哑节点”特性:主要体现在队列初始化时。这个哑节点简化了后续的入队和出队操作,使得 head.next 可以稳定地指向第一个真正等待的线程(如果存在)。
  • head 代表持有锁的节点:当一个线程从等待队列中被唤醒并成功获取锁后,它对应的节点会成为新的 head
  • head 自身永不等待:无论 head 是哑节点还是代表持有锁的节点,它所关联的线程(如果有的话,即非哑节点情况)都不是正在等待锁的线程。
  • 第一个等待者是 head.next:真正排在队列最前面等待获取锁的节点始终是 head.next(如果队列不为空)。

所以,“head到底是哑节点还是一个拿到锁的节点”,这取决于观察的时间点:

  • 初始化时:是哑节点。
  • 有线程持有锁且队列机制运作后:是代表已拿到锁的那个节点。

但最重要的共性是,head 节点本身不参与等待,它是一个“已服务”或“特殊标记”的节点。

isHeldExclusively() 方法

isHeldExclusively() 方法在 AQS 的设计中扮演着一个重要的角色,尤其是在与条件队列 (ConditionObject) 配合使用时。简单来说,isHeldExclusively() 方法的目的是判断当前的同步状态是否被当前线程独占性地持有

以下是 isHeldExclusively() 的几个关键点:

  1. 模板方法AbstractQueuedSynchronizer 本身并没有提供 isHeldExclusively() 的具体实现,而是抛出 UnsupportedOperationException

    AbstractQueuedSynchronizer.java

    // ... existing code ...
    protected boolean isHeldExclusively() {throw new UnsupportedOperationException();
    }
    // ... existing code ...
    

    这意味着它是一个受保护的(protected)模板方法,期望由 AQS 的子类根据其自身的同步语义来重写和实现。子类需要定义“什么是独占持有”以及“如何判断当前线程是否独占持有”。

  2. ConditionObject 的核心依赖isHeldExclusively() 最主要的使用者是 AQS 内部定义的 ConditionObject 类。ConditionObject 实现了 java.util.concurrent.locks.Condition 接口,用于线程间的协作(如 await()signal()signalAll())。 根据 Condition 接口的规范,当一个线程调用一个 Condition 对象的 await() (等待) 或 signal() (通知) 相关方法时,该线程必须已经持有与此 Condition 相关联的锁。 ConditionObject 内部在执行这些操作之前,会调用其外部 AQS 实例的 isHeldExclusively() 方法来检查当前线程是否确实持有了锁。如果 isHeldExclusively() 返回 falseConditionObject 就会抛出 IllegalMonitorStateException,这与使用 Object.wait()/notify() 时不在 synchronized 块中会抛出此异常的行为是一致的。

  3. 子类如何实现: 通常,支持独占模式的 AQS 子类(如 ReentrantLock.Sync)会这样实现 isHeldExclusively()

    ReentrantLock.java

    // ... existing code .../*** Performs lock. The main reason for subclassing is to allow exporting* of non-public methods to public lock methods.*/abstract static class Sync extends AbstractQueuedSynchronizer {private static final long serialVersionUID = -5179523762034025860L;/*** Performs {@link Lock#lock}. The main reason for subclassing* is to allow fast path for non-fair version.*/abstract void lock();/*** 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) // overflowthrow new Error("Maximum lock count exceeded");setState(nextc);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;}protected final boolean isHeldExclusively() {// While we must in general read state before owner,// we don't need to do so to check if current thread is ownerreturn getExclusiveOwnerThread() == Thread.currentThread();}final ConditionObject newCondition() {return new ConditionObject();}// Methods relayed from outer classfinal Thread getOwner() {return getState() == 0 ? null : getExclusiveOwnerThread();}final int getHoldCount() {return isHeldExclusively() ? getState() : 0;}final boolean isLocked() {return getState() != 0;}/*** Reconstitutes the instance from a stream (that is, deserializes it).*/private void readObject(java.io.ObjectInputStream s)throws java.io.IOException, java.lang.ClassNotFoundException {s.defaultReadObject();setState(0); // reset to unlocked state}}
    // ... existing code ...
    

    它通过 getExclusiveOwnerThread() (继承自 AbstractOwnableSynchronizer,AQS 的父类) 获取当前持有锁的线程,并与 Thread.currentThread() 进行比较。

    在 AQS 文档的示例 Mutex 中,也是类似的实现:

    AbstractQueuedSynchronizer.java

    // ... existing code ...* class Mutex implements Lock, java.io.Serializable {**   // Our internal helper class*   private static class Sync extends AbstractQueuedSynchronizer {*     // Acquires the lock if state is zero*     public boolean tryAcquire(int acquires) {*       assert acquires == 1; // Otherwise unused*       if (compareAndSetState(0, 1)) {*         setExclusiveOwnerThread(Thread.currentThread());*         return true;*       }*       return false;*     }**     // Releases the lock by setting state to zero*     protected boolean tryRelease(int releases) {*       assert releases == 1; // Otherwise unused*       if (!isHeldExclusively())*         throw new IllegalMonitorStateException();*       setExclusiveOwnerThread(null);*       setState(0);*       return true;*     }**     // Reports whether in locked state*     public boolean isLocked() {*       return getState() != 0;*     }**     public boolean isHeldExclusively() {*       // a data race, but safe due to out-of-thin-air guarantees*       return getExclusiveOwnerThread() == Thread.currentThread();*     }**     // Provides a Condition*     public Condition newCondition() {*       return new ConditionObject();*     }
    // ... existing code ...
    
  4.  isHeldExclusively() 的主要作用域是在 AQS 之上构建更高级同步工具(特别是 Condition)时,用来确保这些工具的使用者遵守了“必须持有锁”的前置条件。所以,它的调用通常发生在 ConditionObject 内部,而不是 AQS 的主同步路径上。

总结一下: isHeldExclusively() 是 AQS 框架提供给子类的一个钩子方法,用于判断当前线程是否独占了同步器。它对于 ConditionObject 的正确运作至关重要,确保了在使用条件变量时,线程总是持有相关的锁。如果实现自己的基于 AQS 的独占锁,并且希望支持 Condition,那么正确地重写 isHeldExclusively() 是必不可少的。

条件变量 (ConditionObject)

ConditionObject 的 await 和 signal 机制与 AQS 的主队列协同工作:

  • await():
    1. 检查当前线程是否被中断,如果是则抛出 InterruptedException
    2. 创建一个新的 ConditionNode
    3. 将该 ConditionNode 加入到此 ConditionObject 内部的条件等待队列的尾部。
    4. 完全释放当前线程持有的锁(通过调用 AQS 的 release 方法,并传入当前 state)。
    5. 阻塞当前线程 (LockSupport.park),直到被 signalsignalAll 或中断。
    6. 当线程被唤醒后:
      • 它会尝试重新获取锁(通过调用 AQS 的 acquire 方法)。在获取锁之前,它会被从条件队列转移到 AQS 的主同步队列中排队。
      • 如果线程在等待期间被中断,await 方法会抛出 InterruptedException (在重新获取锁之后)。
  • signal():
    1. 从条件等待队列的头部取出一个 ConditionNode (如果队列不为空)。
    2. 将该节点从条件队列转移到 AQS 的主同步队列的尾部,等待获取锁。
    3. 如果转移成功,并且节点的状态允许,会尝试唤醒该节点对应的线程(通过 LockSupport.unpark)。被唤醒的线程接下来会尝试获取锁。

awaitNanosawaitUntilawaitUninterruptibly 等方法与 await 类似,但增加了超时和不可中断的逻辑。

AbstractQueuedSynchronizer.java

// ... existing code ...public final void await() throws InterruptedException {if (Thread.interrupted()) // 1. 检查中断throw new InterruptedException();ConditionNode node = newConditionNode(); // 2. 创建条件节点if (node == null)return;int savedState = enableWait(node); // 3. 加入条件队列, 4. 释放锁 (enableWait内部调用addWaiter和release)LockSupport.setCurrentBlocker(this); // for back-compatibilityboolean interrupted = false, cancelled = false, rejected = false;while (!canReacquire(node)) { // 5. 循环检查是否可以重新获取锁 (即是否被signal并转移到主队列且轮到它)if (interrupted |= Thread.interrupted()) { // 检查中断if (cancelled = (node.getAndUnsetStatus(COND) & COND) != 0)break;              // else interrupted after signal} else if ((node.status & COND) != 0) { // 如果仍在条件队列中等待try {if (rejected)node.block(); // 使用 LockSupport.park() 阻塞elseForkJoinPool.managedBlock(node); // 尝试使用 ForkJoinPool 的阻塞机制} catch (RejectedExecutionException ex) {rejected = true;} catch (InterruptedException ie) { // 如果在 park/managedBlock 时被中断interrupted = true;if (cancelled = (node.getAndUnsetStatus(COND) & COND) != 0)break;}} else // 从 park/managedBlock 返回,但不在条件队列了 (可能被 signal)Thread.onSpinWait();    // awoke while enqueuing}LockSupport.setCurrentBlocker(null);node.clearStatus();reacquire(node, savedState); // 6. 重新获取锁if (cancelled)unlinkCancelledWaiters(node);if (interrupted) // 6. 如果曾被中断,抛出异常throw new InterruptedException();}public final void signal() {if (!isHeldExclusively()) // 必须持有锁才能signalthrow new IllegalMonitorStateException();ConditionNode first = firstWaiter;if (first != null && (first.getAndUnsetStatus(COND) & COND) != 0)doSignal(first); // 1 & 2 & 3. 从条件队列取节点,转移到主队列并尝试唤醒}
// ... existing code ...

doSignal(ConditionNode first, boolean all) 方法分析

这个方法的主要职责是将一个或所有在条件队列中等待的节点(ConditionNode)转移到同步队列(AQS的等待队列)中,让它们有机会重新竞争锁。

AbstractQueuedSynchronizer.java

// ... existing code ...public class ConditionObject implements Condition, java.io.Serializable {
// ... existing code .../*** Removes and transfers one or all waiters to sync queue.*/private void doSignal(ConditionNode first, boolean all) {while (first != null) { // 循环处理条件队列中的节点ConditionNode next = first.nextWaiter; // 保存下一个等待者// 更新条件队列的头节点// 如果next为null,表示当前first是最后一个节点,清空条件队列if ((firstWaiter = next) == null)lastWaiter = null;else// 否则,断开当前节点与后续节点的链接(帮助GC)first.nextWaiter = null; // GC assistance// 尝试将节点状态从COND迁移,如果成功(表示节点确实在等待条件)// getAndUnsetStatus(COND) 会原子地清除COND位并返回旧的状态值// (oldStatus & COND) != 0 确保该节点确实是因为在条件上等待而被唤醒if ((first.getAndUnsetStatus(COND) & COND) != 0) {enqueue(first); // 将节点加入到AQS的同步队列中if (!all) // 如果不是signalAll操作(即只是signal操作),只处理第一个节点就退出break;}// 处理条件队列中的下一个节点first = next;}}
// ... existing code ...}
// ... existing code ...

方法逻辑总结:

  1. 遍历节点: 方法通过一个 while 循环遍历从 first 开始的条件队列节点。
  2. 更新条件队列:
    • 保存当前节点 first 的 nextWaiter
    • 将条件队列的头(firstWaiter)指向 next。如果 next 为 null,说明条件队列变空,同时将尾(lastWaiter)也设为 null
    • 将当前节点 first 的 nextWaiter 设为 null,这有助于垃圾回收,因为它断开了该节点在条件队列中的前向链接。
  3. 状态检查与入队:
    • first.getAndUnsetStatus(COND): 这是一个关键的原子操作。它会尝试清除节点状态 status 中的 COND 位 (值为2)。如果 COND 位之前是设置的(即 (旧状态 & COND) != 0),说明这个节点确实是在等待这个条件。
    • 如果节点确实在等待条件,则调用 enqueue(first) 方法,将这个 ConditionNode 加入到 AQS 的主同步队列中。
  4. signal vs signalAll:
    • 如果 all 参数为 false(对应 signal() 操作),则只处理第一个满足条件的节点,然后通过 break 退出循环。
    • 如果 all 参数为 true(对应 signalAll() 操作),则会继续循环,处理条件队列中所有满足条件的节点。
  5. 迭代: 将 first 指向之前保存的 next,继续处理下一个节点。

enqueue(ConditionNode node) 方法分析

这个方法负责将一个(通常是从条件队列转移过来的)ConditionNode 添加到 AQS 的同步队列的尾部。

AbstractQueuedSynchronizer.java

// ... existing code .../*** Enqueues the node unless null. (Currently used only for* ConditionNodes; other cases are interleaved with acquires.)*/final void enqueue(ConditionNode node) {if (node != null) { // 确保节点不为nullboolean unpark = false; //标记是否需要唤醒节点对应的线程for (Node t;;) { // 无限循环,直到节点成功入队或发生OOME// 获取当前的尾节点。如果尾节点为null,说明同步队列还未初始化// 尝试初始化头节点,如果初始化失败(可能因为OOME),则t也为nullif ((t = tail) == null && (t = tryInitializeHead()) == null) {unpark = true;             // 唤醒线程让其自旋等待OOME恢复break;}// 设置当前节点的前驱为旧的尾节点t。// setPrevRelaxed避免了不必要的内存屏障,依赖后续的CAS操作保证可见性node.setPrevRelaxed(t);        // avoid unnecessary fence// 尝试通过CAS将当前节点设置为新的尾节点if (casTail(t, node)) {// CAS成功,旧的尾节点t的next指向当前节点nodet.next = node;// 如果旧的尾节点t的状态是负数(通常表示CANCELLED)// 那么当前节点入队后可能需要被唤醒以进行队列清理或尝试获取锁if (t.status < 0)          // wake up to clean linkunpark = true;break; // 成功入队,跳出循环}// CAS失败,意味着有其他线程修改了tail,循环重试}// 如果需要唤醒if (unpark)LockSupport.unpark(node.waiter); // 唤醒与此节点关联的线程}}
// ... existing code ...

方法逻辑总结:

  1. 非空检查: 首先检查传入的 node 是否为 null
  2. 循环入队 (CAS): 使用一个无限循环和 CAS (compareAndSetTail) 来保证节点原子性地添加到队尾。
    • 获取尾节点/初始化:
      • t = tail: 获取当前 AQS 队列的尾节点。
      • 如果 tail 是 null(队列未初始化),则调用 tryInitializeHead()。这个方法会尝试创建一个虚拟头节点,并将 head 和 tail 都指向它。
      • 如果 tryInitializeHead() 也返回 null(通常是因为在创建节点时发生 OutOfMemoryError),则将 unpark 标记为 true 并跳出循环。这意味着节点入队失败,但其线程会被唤醒,可能会在 acquireOnOOME 中进行自旋等待。
    • 链接节点:
      • node.setPrevRelaxed(t): 将新节点的 prev 指针指向当前的尾节点 tRelaxed 后缀意味着这是一个“松散”的写操作,不保证立即对其他线程可见,依赖于后续的 casTail 操作来确保正确的内存顺序。
    • CAS 更新尾节点:
      • casTail(t, node): 尝试原子地将 AQS 队列的 tail 从 t 更新为 node
      • 成功: 如果 CAS 成功,说明 node 已经成为新的尾节点。
        • t.next = node: 将旧尾节点 t 的 next 指针指向新节点 node,完成双向链表的链接。
        • if (t.status < 0): 检查旧尾节点 t 的状态。如果状态是负值(例如 Node.CANCELLED),说明前一个节点被取消了。这种情况下,新入队的节点可能需要被唤醒来帮助清理队列或者因为它可能是新的有效头节点。因此,设置 unpark = true
        • break: 成功入队,跳出循环。
      • 失败: 如果 CAS 失败,说明在当前线程尝试更新 tail 时,有其他线程已经修改了 tail。循环会继续,重新获取 tail 并重试。
  3. 唤醒线程:
    • if (unpark): 如果 unpark 标记为 true(因为 OOME 导致初始化失败,或者前一个节点被取消),则调用 LockSupport.unpark(node.waiter) 来唤醒与 node 关联的线程。

两者如何协同工作

  1. 当一个线程在某个 ConditionObject 上调用 signal() 或 signalAll() 时,会执行 doSignal 方法。
  2. doSignal 会从条件队列中选择一个或多个 ConditionNode
  3. 对于每个被选中的 ConditionNodedoSignal 会首先原子地更新其状态(清除 COND 位),确认它确实在等待该条件。
  4. 然后,doSignal 调用 enqueue(node),将被选中的 ConditionNode 传递给它。
  5. enqueue 方法负责将这个 ConditionNode 安全地、原子地添加到 AQS 的主同步队列的尾部。
  6. 在 enqueue 的过程中,如果遇到队列未初始化的情况,它会尝试初始化队列。如果遇到前一个节点被取消的情况,或者初始化时发生 OOME,它会标记需要唤醒当前入队节点的线程。
  7. 最终,被转移到同步队列的节点(线程)会在合适的时机(通常是被前一个节点释放锁后唤醒,或者因为 unpark 标志被设置而唤醒)尝试重新获取锁。

总的来说,doSignal 负责决策哪些等待条件的线程需要被唤醒并转移,而 enqueue 负责将这些线程对应的节点实际地、安全地放入 AQS 的等待队列中,为它们后续竞争锁做准备。unpark 逻辑确保了在特定情况下(如前驱节点取消或队列初始化问题)线程能够被及时唤醒。

设计与实现的关键点

  • 原子操作: AQS 广泛使用 sun.misc.Unsafe (在较新 JDK 中通过 jdk.internal.misc.Unsafe) 提供的 CAS 操作来保证对 stateheadtail 以及 Node 中某些字段的原子更新,从而避免了使用传统锁带来的开销和复杂性。
  • CLH 队列变体: 使用双向链表,方便节点取消时的处理。prev 指针用于处理取消,next 指针主要用于唤醒后继。
  • 惰性初始化head 和 tail 是在第一次发生竞争时才初始化的,减少了无竞争情况下的开销。
  • 状态驱动Node 的 status 字段非常关键,它驱动了线程的阻塞和唤醒逻辑。
  • 公平性与非公平性:
    • AQS 本身不强制公平性。默认情况下,新请求锁的线程可能会“插队”(barging),直接尝试获取锁,而不是先入队。这通常能带来更高的吞吐量。
    • 子类可以通过在 tryAcquire 或 tryAcquireShared 方法中检查 hasQueuedPredecessors() (判断队列中是否有等待时间更长的线程) 来实现公平锁。
  • 中断与超时acquireInterruptibly 和 tryAcquireNanos 方法提供了可中断和可超时的获取锁的逻辑。acquire 方法的核心循环会检查线程的中断状态,并在适当的时候处理。
  • 内存可见性stateheadtail 以及 Node 中的关键字段(如 statusprevnext)都用 volatile 修饰,或者通过 CAS 操作的内存语义来保证跨线程的可见性。
  • LockSupport: 线程的阻塞和唤醒依赖于 LockSupport.park() 和 LockSupport.unpark()。这比传统的 Object.wait() 和 Object.notify() 更灵活,因为它们不需要获取对象的监视器锁,并且 unpark 可以先于 park 调用。

总结:

AbstractQueuedSynchronizer 是一个强大且灵活的同步框架。它通过一个原子整数 state 和一个 FIFO 等待队列,为构建各种同步器提供了一个通用的、高性能的基础。理解 AQS 的工作原理对于深入理解 Java 并发包中的锁和其他同步工具至关重要。开发者可以通过继承 AQS 并实现其定义的几个核心 try* 方法,来定制满足特定需求的同步逻辑。

http://www.xdnf.cn/news/10041.html

相关文章:

  • 宝塔部署 Vue + NestJS 全栈项目
  • 单片机寄存器的四种主要类型!
  • AWS Transit Gateway实战:构建DMZ隔离架构,实现可控的网络互通
  • 模块化设计,static和extern(面试题常见)
  • 2025.5.30工作总结
  • ubuntu20.04安装教程(图文详解)
  • LangChain-结合魔塔社区modelscope的embeddings实现搜索
  • Java八股文——Java基础「概念篇」
  • azure web app创建分步指南
  • 从虚拟化到云原生与Serverless
  • CSS 渐变完全指南:从基础概念到实战案例(线性渐变/径向渐变/重复渐变)
  • 异步并发控制代码详细分析
  • (c++)string的模拟实现
  • 【Office】Excel两列数据比较方法总结
  • 基于大模型预测的FicatIII-IV期股骨头坏死综合治疗研究报告
  • 多模态大语言模型arxiv论文略读(100)
  • LNMP环境中php7.2升级到php7.4
  • Android Native 之 adbd进程分析
  • 视频监控汇聚平台EasyCVR安防小知识:如何通过视频融合平台解决信息孤岛问题?
  • @Pushgateway 数据自动清理
  • 碰一碰发视频系统--基于H5场景开发
  • 选择if day5
  • QPS 和 TPS 详解
  • 竞争加剧,美团的战略升维:反内卷、科技与全球化
  • C++ 游戏开发详细流程
  • 大规模JSON反序列化性能优化实战:Jackson vs FastJSON深度对比与定制化改造
  • Elasticsearch 分析器介绍
  • Camera相机人脸识别系列专题分析之六:MTK ISP6S平台人脸识别fdnode流程FdNodeImp.cpp详解
  • Xamarin劝退之踩坑笔记
  • 苹果签名!