Go back

AQS——JUC所有锁的共同底座

AQS：ReentrantLock、Semaphore、CountDownLatch 的共同底座

一句话结论（30s）

AQS 的本质是一个「抽离了阻塞、排队、唤醒脏活」的抽象基类，ReentrantLock、Semaphore、CountDownLatch 都只是往里面填 state 语义和 tryAcquire/tryRelease 模板方法的不同实现。它的关键设计是 volatile int state 这个「多义词」+ CLH 变体双向队列 + park/unpark 精准唤醒。核心权衡是非公平锁用「新线程直接 CAS 抢锁」换吞吐——因为锁释放后队头线程还在等 OS 调度恢复的微秒级空窗正好被新线程插队利用，代价是牺牲 FIFO 公平性。

核心原理（2min）

主流程：state 记录同步状态（锁的重入次数 / 剩余许可 / 剩余 countDown），子类只定义 tryAcquire/tryRelease 的语义，AQS 负责入队、挂起、唤醒。加锁失败时线程被 CAS 尾插入 CLH 变体队列并 park 挂起；释放时 unpark 队头线程。关键机制两点：一是非公平锁的 lock() 先 compareAndSetState(0, 1) 直接抢，抢不到才 acquire(1) 排队，而公平锁的 tryAcquire 先查 hasQueuedPredecessors()，有前驱就乖乖排队；二是用 LockSupport.park/unpark 而非 wait/notify，因为 unpark 允许在 park 之前调用（许可累积、不丢信号），且能只唤醒指定线程、不惊群。

底层深入（5-10min）

AQS 的 state + CLH 变体队列

// 核心字段：同步状态 + CLH 变体队列的头尾指针（真实源码）
private transient volatile Node head;
private transient volatile Node tail;
private volatile int state;

// Node 状态位（真实源码）
static final int WAITING   = 1;          // must be 1
static final int CANCELLED = 0x80000000; // must be negative
static final int COND      = 2;          // in a condition wait
// CLH 节点：双向链表节点，存线程引用 + 状态位（真实源码）
abstract static class Node {
    volatile Node prev;       // initially attached via casTail
    volatile Node next;       // visibly nonnull when signallable
    Thread waiter;            // visibly nonnull when enqueued
    volatile int status;      // written by owner, atomic bit ops by others

    // methods for atomic operations
    final boolean casPrev(Node c, Node v) {  // for cleanQueue
        return U.weakCompareAndSetReference(this, PREV, c, v);
    }
    final boolean casNext(Node c, Node v) {  // for cleanQueue
        return U.weakCompareAndSetReference(this, NEXT, c, v);
    }
    final int getAndUnsetStatus(int v) {     // for signalling
        return U.getAndBitwiseAndInt(this, STATUS, ~v);
    }
    final void setPrevRelaxed(Node p) {      // for off-queue assignment
        U.putReference(this, PREV, p);
    }
    final void setStatusRelaxed(int s) {     // for off-queue assignment
        U.putInt(this, STATUS, s);
    }
    final void clearStatus() {               // for reducing unneeded signals
        U.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");
}

Node.status 用 volatile int 加 Unsafe 的原子位操作维护：WAITING=1 表示线程需要被 unpark 唤醒，CANCELLED（最高位为负）表示已取消（中断/超时），COND=2 表示正在条件队列上等待。相比旧版的 waitStatus（SIGNAL/CANCELLED/CONDITION/PROPAGATE），新版把语义收敛成三种位，waiter 字段直接存 Thread 引用，节点自身不再包一层。

想一想：为什么要有 CLH 队列，而不是让竞争线程各自无脑 CAS 重试？ 因为无脑 CAS 重试会退化成「惊群 + 饿死」：所有竞争线程反复抢同一把锁，谁抢到完全随机，可能有的线程永远抢不到（不公平），且每次释放都会唤醒一堆线程再互相竞争。CLH 队列把所有失败线程按到达顺序排成 FIFO，释放时只精准唤醒队头（signalNext(head)），既避免惊群又保证先到先得——这就是「排队」比「抢」更可预期的原因。

state 是一个”多义词”：

想一想：state 为什么必须是 volatile 的？ 因为多个线程都要读这个状态位判断「能不能抢锁」，如果 state 不是 volatile，线程 A 释放锁改的 state 值可能一直缓存在自己的 CPU Cache 里，线程 B 永远看不到「锁已经释放了」。volatile 保证 state 的写对其它线程立即可见，再配合 compareAndSetState 的 CAS 做原子更新——两者一个管可见性、一个管原子性，缺一不可。

AQS 把所有同步器的共性抽离（阻塞队列、state、CAS），子类只需定义 tryAcquire/tryRelease 这几个模板方法，AQS 负责队列管理、线程挂起/唤醒。真实源码里这两个模板方法的默认实现如下：

// 模板方法：子类定义「什么算获得成功」（真实源码，默认直接抛异常）
protected boolean tryAcquire(int arg) {
    throw new UnsupportedOperationException();
}

// 原子更新 state（真实源码）
protected final boolean compareAndSetState(int expect, int update) {
    return U.compareAndSetInt(this, STATE, expect, update);
}

tryAcquire 默认直接抛 UnsupportedOperationException——AQS 本身不定义「获得锁」的语义，只提供 compareAndSetState 这个 CAS 基座，把 state 的读写原子化。子类（如 ReentrantLock 的 NonfairSync/FairSync）必须覆写 tryAcquire/tryRelease，把 state 翻译成自己的语义（0/1 锁、重入次数、许可数、countDown 剩余数）。这就是模板方法模式：AQS 提供不变的「排队/阻塞/唤醒」骨架，子类只填「什么时候算抢到/抢失败」这一步。

公平锁 vs 非公平锁

**非公平锁（默认）**的 lock()：

// NonfairSync.lock()
final void lock() {
    if (compareAndSetState(0, 1))     // 直接抢！不管队列有没有排队者
        setExclusiveOwnerThread(current);
    else
        acquire(1);                    // 抢不到才排队
}

公平锁的 tryAcquire()：

// FairSync.tryAcquire()
protected final boolean tryAcquire(int acquires) {
    if (hasQueuedPredecessors())       // 队列有等待者 → 不抢 → 乖乖排队
        return false;
    if (compareAndSetState(0, acquires)) { ... }
}

非公平锁吞吐更高的原因：锁刚被释放 → 队头的等待线程被 unpark 后还在等 OS 调度恢复（微秒级延迟）→ 此时新来的线程直接 CAS 抢锁成功并开始执行。这微秒级的”插队”减少了 CPU 等待 OS 调度队头线程的空转时间，整体吞吐更高。

公平锁保证 FIFO，队头线程永远先获取锁 → 上下文切换次数少（队头线程被 park→unpark 后直接获取，不需要和新来的线程竞争）→ 但每次锁释放都必须等 OS 调度队头线程，这中间有微秒级的锁空闲窗口。

想一想：非公平锁插队到底是「赚」还是「亏」？ 从吞吐看是赚——它利用队头线程被 unpark 后等 OS 调度恢复的那段微秒级空窗，让新线程直接 CAS 抢到锁开始干活，CPU 不用干等；但从公平看是亏——后到的线程可能反复插队，让队里的线程迟迟拿不到锁。所以默认用非公平锁是「吞吐优先」的取舍，只有对延迟敏感、要求严格 FIFO 的场景才显式选公平锁。

上面两段 NonfairSync.lock() / FairSync.tryAcquire() 其实来自 ReentrantLock 内部的 Sync 子类，不是 AQS.java 本身——它们最终都汇聚到下面 AQS 的 acquire(1) / release(1)。

真实源码：acquire 主循环、入队与 release

旧版 AQS 把加锁拆成 tryAcquire + addWaiter（建节点）+ enq（CAS 尾插）+ acquireQueued（自旋/挂起）+ shouldParkAfterFailedAcquire/parkAndCheckInterrupt 一连串方法。当前 JDK 把这些全部合并进一个 acquire(Node, int, ...) 大循环，enq/acquireQueued 方法已不存在。真实的入口与主循环如下：

public final void acquire(int arg) {
    if (!tryAcquire(arg))
        acquire(null, arg, false, false, false, 0L);
}
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 thread
    boolean 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))) {
            if (pred.status < 0) {
                cleanQueue();           // predecessor cancelled
                continue;
            } else if (pred.prev == null) {
                Thread.onSpinWait();    // ensure serialization
                continue;
            }
        }
        if (first || pred == null) {
            boolean acquired;
            try {
                if (shared)
                    acquired = (tryAcquireShared(arg) >= 0);
                else
                    acquired = tryAcquire(arg);
            } catch (Throwable ex) {
                cancelAcquire(node, interrupted, false);
                throw ex;
            }
            if (acquired) {
                if (first) {
                    node.prev = null;
                    head = node;
                    pred.next = null;
                    node.waiter = null;
                    if (shared)
                        signalNextIfShared(node);
                    if (interrupted)
                        current.interrupt();
                }
                return 1;
            }
        }
        Node t;
        if ((t = tail) == null) {           // initialize queue
            if (tryInitializeHead() == null)
                return acquireOnOOME(shared, arg);
        } else if (node == null) {          // allocate; retry before enqueue
            try {
                node = (shared) ? new SharedNode() : new ExclusiveNode();
            } catch (OutOfMemoryError oome) {
                return acquireOnOOME(shared, arg);
            }
        } else if (pred == null) {          // try to enqueue
            node.waiter = current;
            node.setPrevRelaxed(t);         // avoid unnecessary fence
            if (!casTail(t, node))
                node.setPrevRelaxed(null);  // back out
            else
                t.next = node;
        } else if (first && spins != 0) {
            --spins;                        // reduce unfairness on rewaits
            Thread.onSpinWait();
        } else if (node.status == 0) {
            node.status = WAITING;          // enable signal and recheck
        } 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);
                else
                    break;
            } catch (Error | RuntimeException ex) {
                cancelAcquire(node, interrupted, interruptible); // cancel & rethrow
                throw ex;
            }
            node.clearStatus();
            if ((interrupted |= Thread.interrupted()) && interruptible)
                break;
        }
    }
    return cancelAcquire(node, interrupted, interruptible);
}

这个循环把旧版 enq/acquireQueued 的动作全摊平在一个 for(;;) 里：先判断自己是否已是队头（first），是就调 tryAcquire 抢；抢不到再按「初始化队列 → 分配节点 → CAS 尾插（旧 enq）→ 自旋 → 置 WAITING → park」的顺序逐级推进。其中「try to enqueue」分支就是旧 enq 的现代替身——casTail(t, node) 原子尾插成功后，前驱的 next 再指向自己，失败就 setPrevRelaxed(null) 回退重试。真正阻塞用的是 LockSupport.park(this)，与下面 release 的 unpark 精准配对。

旧 enq 里「懒初始化 dummy head」的职责，现在由 tryInitializeHead 承担，真实源码如下：

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))
                return tail = h;
        }
    }
}

队列是「首次竞争才建」的：只有 tail == null 且 CAS 抢锁失败时才新建 dummy head，把 head/tail 同时指向它。这样无竞争场景完全不会产生任何节点对象，避免无谓的 GC 压力。多个线程同时初始化时靠 compareAndSetReference(HEAD, null, h) 保证只有一个成功，失败的 onSpinWait 自旋等待。

释放端的 release 与唤醒后继的 signalNext，真实源码如下：

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        signalNext(head);
        return true;
    }
    return false;
}
private static void signalNext(Node h) {
    Node s;
    if (h != null && (s = h.next) != null && s.status != 0) {
        s.getAndUnsetStatus(WAITING);
        LockSupport.unpark(s.waiter);
    }
}

release 先让子类 tryRelease 把 state 减到位（重入锁要减到 0 才返回 true），再 signalNext(head) 唤醒头结点的直接后继。signalNext 先 getAndUnsetStatus(WAITING) 原子清除后继的 WAITING 位，再 unpark 它的线程——先清位再唤醒是为了消除「线程还没 park 但 unpark 先到」的竞态：被唤醒的线程在 acquire 循环里会先 tryAcquire 再决定要不要 park，许可不会丢。

park/unpark 为什么比 wait/notify 更适合

Object.wait/notify 在 AQS 场景下有致命弱点：

  1. 必须持 Monitor 才能调用：AQS 入队时通常没有锁（CAS 失败才入队），wait() 报错
  2. notify 在 wait 之前调用信号丢失：AQS 的 release() → unpark(next.thread) 之后，新的线程才尝试 acquire() → park()。如果用 notify/wait，notify 先到达，wait 永远等不到
  3. notifyAll 惊群：所有等待线程都被唤醒，只有一个能获取锁，其他全被重新阻塞

LockSupport.unpark(thread) 在 park() 之前调用 → 许可累积 → 下一次 park() 直接消费许可，不阻塞。AQS 用这个特性精准唤醒 CLH 队列的队头线程，不惊群。

想一想：为什么 unpark 可以在 park 之前调用却不丢信号？ 因为 LockSupport 内部维护的是「许可」而非「通知」——unpark 只是把许可置为 1，之后线程调用 park 时发现已有许可就直接消费并返回，不会阻塞。这正好化解了 notify 先于 wait 到达时信号丢失的竞态，也是 AQS 敢让 release 先 unpark、线程后 park 的底气。

总结

同步器state 含义典型方法
ReentrantLock0=无锁, ≥1=持锁/重入lock, unlock
Semaphore剩余许可数acquire, release
CountDownLatch剩余 countDown 次数countDown, await

AQS 的设计精髓：把阻塞、排队、唤醒这些所有同步器都需要的”脏活”抽离到一个抽象类中，子类只需定义”什么是可获得/什么是可获得失败”的语义。

章末提问

  1. AQS 为什么能成为 ReentrantLock、Semaphore、CountDownLatch 的共同底座？ 结论：因为它把「阻塞、排队、唤醒」这些所有同步器都要的脏活抽离成模板方法，子类只覆写 tryAcquire/tryRelease 给 state 定义不同语义。因为加锁的公共流程是固定的，只有「什么算抢到」是各同步器独有的，模板方法模式正好切出这个变点。

  2. state 为什么必须用 volatile + CAS，而不是普通 int？ 结论：因为 volatile 保证 state 的写对其它线程可见，CAS（compareAndSetState）保证读-改-写的原子性，两者缺一不可。因为多线程并发改同一个计数，光可见不够（会丢更新）、光原子不够（看不到最新值）。

  3. 为什么默认用非公平锁？它为什么吞吐更高？ 结论：非公平锁让新线程直接 CAS 抢锁，利用队头线程被 unpark 后等 OS 调度恢复的微秒级空窗插队，减少 CPU 空转。因为锁刚释放到队头线程真正跑起来之间有空档，新线程插队正好填上这段空档。

  4. 为什么 AQS 用 LockSupport.park/unpark 而不是 wait/notify？ 结论：因为 unpark 可以在 park 之前调用（许可累积、信号不丢）、能精准唤醒指定线程不惊群、且调用时无需持有 Monitor。因为 AQS 入队时通常没持锁，wait/notify 的三条限制它都满足不了。

  5. CLH 变体队列解决了什么问题？ 结论：把竞争失败的线程按 FIFO 排队、释放时只精准唤醒队头，避免无脑 CAS 重试的惊群和饿死。因为队头节点就是「先到先得」的载体，signalNext(head) 一次只唤醒一个，谁先来谁先拿。

Share this post on:
Share this post via WhatsAppShare this post on FacebookShare this post on XShare this post via TelegramShare this post on PinterestShare this post via email
Previous Post
CompletableFuture——异步编程的链式编排
Next Post
Java线上排查——从top到Arthas的完整工具链