1. 理解AQS的本质第一次接触AQS时我盯着这个缩写看了半天——AbstractQueuedSynchronizer抽象队列同步器。名字听起来就很唬人但当我真正理解它的设计哲学后才发现这简直是Java并发编程中的瑞士军刀。AQS本质上是一个构建锁和同步器的框架它用了一个int成员变量表示同步状态通过内置的FIFO队列来完成资源获取线程的排队工作。重要提示AQS不是直接给开发者用的工具类而是给那些需要实现自己同步原语的大神们准备的底层框架。我们日常用的ReentrantLock、CountDownLatch等其实都是基于AQS实现的。AQS最精妙的地方在于它采用了模板方法模式。它定义了acquire和release这样的模板方法但把具体的资源获取和释放逻辑留给子类实现。这种设计让AQS既能保持核心排队机制的稳定性又能灵活支持各种同步场景。2. AQS的核心机制解析2.1 同步状态与CAS操作AQS内部维护了一个volatile int类型的state变量这个变量就是同步状态的核心。不同的同步器对state的解释不同在ReentrantLock中state表示锁的持有计数在Semaphore中state表示可用许可数在CountDownLatch中state表示计数器值所有对state的修改都通过CASCompare-And-Swap操作完成这是AQS能够实现无锁并发的基础。在JDK中这个操作通过Unsafe类实现protected final boolean compareAndSetState(int expect, int update) { return unsafe.compareAndSwapInt(this, stateOffset, expect, update); }2.2 等待队列的实现当线程获取资源失败时AQS会将线程封装成一个Node节点加入CLH队列一种改进的CLH锁队列。这个队列是AQS能够实现公平性的关键static final class Node { volatile int waitStatus; volatile Node prev; volatile Node next; volatile Thread thread; Node nextWaiter; // ... }每个Node节点保存了线程引用、前后节点指针以及等待状态。队列的入队和出队操作同样通过CAS保证线程安全。2.3 独占与共享模式AQS支持两种同步模式独占模式Exclusive同一时刻只有一个线程能获取资源如ReentrantLock共享模式Shared多个线程可以同时获取资源如Semaphore这两种模式通过不同的acquire/release方法实现acquire/release独占模式acquireShared/releaseShared共享模式3. 基于AQS实现自定义同步器3.1 实现一个简单的互斥锁让我们通过实现一个简单的不可重入互斥锁来理解AQS的使用public class MyMutex { private static class Sync extends AbstractQueuedSynchronizer { Override protected boolean tryAcquire(int arg) { return compareAndSetState(0, 1); } Override protected boolean tryRelease(int arg) { setState(0); return true; } Override protected boolean isHeldExclusively() { return getState() 1; } } private final Sync sync new Sync(); public void lock() { sync.acquire(1); } public void unlock() { sync.release(1); } }这个实现虽然简单但包含了AQS使用的核心要点继承AQS并实现tryAcquire/tryRelease方法使用state表示锁状态0表示未锁定1表示锁定通过acquire/release方法提供外部接口3.2 实现一个简单的CountDownLatch再来看看如何实现一个简单的CountDownLatchpublic class MyCountDownLatch { private static class Sync extends AbstractQueuedSynchronizer { Sync(int count) { setState(count); } Override protected int tryAcquireShared(int acquires) { return (getState() 0) ? 1 : -1; } Override protected boolean tryReleaseShared(int releases) { for (;;) { int c getState(); if (c 0) return false; int nextc c - 1; if (compareAndSetState(c, nextc)) return nextc 0; } } } private final Sync sync; public MyCountDownLatch(int count) { this.sync new Sync(count); } public void await() throws InterruptedException { sync.acquireSharedInterruptibly(1); } public void countDown() { sync.releaseShared(1); } }这个实现展示了AQS共享模式的使用关键点在于tryAcquireShared返回负值表示获取失败tryReleaseShared通过CAS循环减少计数器当计数器归零时所有等待线程会被唤醒4. AQS在JDK中的应用实例4.1 ReentrantLock的实现ReentrantLock是AQS最经典的用法之一。它的内部类Sync继承自AQS并实现了可重入特性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) // overflow throw new Error(Maximum lock count exceeded); setState(nextc); return true; } return false; }这段代码展示了可重入锁的关键逻辑如果锁未被持有state0尝试获取如果锁已被当前线程持有增加重入计数释放时相应减少计数归零时完全释放4.2 Semaphore的实现Semaphore使用AQS的共享模式来控制资源访问protected int tryAcquireShared(int acquires) { for (;;) { int available getState(); int remaining available - acquires; if (remaining 0 || compareAndSetState(available, remaining)) return remaining; } }Semaphore的tryAcquireShared实现展示了计算剩余许可数如果不足直接返回负值否则尝试CAS更新状态5. AQS性能优化技巧5.1 减少CAS竞争在高并发场景下AQS的CAS操作可能成为瓶颈。我们可以通过以下方式优化使用自旋等待在入队前短暂自旋可能避免队列操作减少state修改频率比如在读写锁中使用位运算分离读写状态使用ThreadLocal缓存对于可重入锁可以缓存部分状态5.2 队列优化AQS的CLH队列可以针对特定场景优化对于短时任务可以设置自旋超时对于读多写少的场景可以偏向读锁使用条件队列时注意避免虚假唤醒6. AQS常见问题排查6.1 死锁问题虽然AQS本身不会导致死锁但使用不当仍然可能发生// 错误示例嵌套锁可能导致死锁 public void transfer(MyMutex lock1, MyMutex lock2) { lock1.lock(); lock2.lock(); try { // 转账操作 } finally { lock2.unlock(); lock1.unlock(); } }解决方案按固定顺序获取锁使用tryLock设置超时使用更高级别的并发工具6.2 性能问题AQS在高竞争场景下可能出现性能问题表现为CAS操作失败率高队列过长导致上下文切换频繁解决方法减小锁粒度使用读写锁分离读写操作考虑无锁算法替代方案7. AQS最佳实践7.1 正确实现tryAcquire实现tryAcquire时需要注意必须保证线程安全考虑可重入性需求正确处理中断protected boolean tryAcquire(int arg) { Thread current Thread.currentThread(); int c getState(); if (c 0) { if (!hasQueuedPredecessors() // 公平性检查 compareAndSetState(0, arg)) { setExclusiveOwnerThread(current); return true; } } else if (current getExclusiveOwnerThread()) { int nextc c arg; if (nextc 0) throw new Error(Maximum lock count exceeded); setState(nextc); return true; } return false; }7.2 合理使用条件队列AQS的条件队列ConditionObject提供了更灵活的等待/通知机制public class BoundedBuffer { private final Lock lock new ReentrantLock(); private final Condition notFull lock.newCondition(); private final Condition notEmpty lock.newCondition(); public void put(Object x) throws InterruptedException { lock.lock(); try { while (count items.length) notFull.await(); items[putPtr] x; if (putPtr items.length) putPtr 0; count; notEmpty.signal(); } finally { lock.unlock(); } } // 类似的take方法... }使用条件队列时要注意总是在循环中检查条件先获取锁再调用await/signal使用多个Condition分离不同的等待条件8. AQS源码深度解析8.1 acquireQueued方法剖析这是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)) { setHead(node); p.next null; // help GC failed false; return interrupted; } if (shouldParkAfterFailedAcquire(p, node) parkAndCheckInterrupt()) interrupted true; } } finally { if (failed) cancelAcquire(node); } }这个方法展示了只有前驱是头节点时才尝试获取公平性获取失败后检查是否需要阻塞处理中断标志最终处理失败情况8.2 取消机制分析当线程在等待过程中被中断或超时需要取消获取操作private void cancelAcquire(Node node) { if (node null) return; node.thread null; // 跳过已取消的前驱 Node pred node.prev; while (pred.waitStatus 0) node.prev pred pred.prev; Node predNext pred.next; node.waitStatus Node.CANCELLED; // 如果是尾节点直接移除 if (node tail compareAndSetTail(node, pred)) { compareAndSetNext(pred, predNext, null); } else { // 否则唤醒后继节点 int ws; if (pred ! head ((ws pred.waitStatus) Node.SIGNAL || (ws 0 compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) pred.thread ! null) { Node next node.next; if (next ! null next.waitStatus 0) compareAndSetNext(pred, predNext, next); } else { unparkSuccessor(node); } node.next node; // help GC } }取消操作需要考虑清理线程引用跳过已取消的前驱处理不同位置的节点必要时唤醒后继9. AQS扩展与变种9.1 响应中断的获取操作AQS提供了多种获取方法应对不同需求acquire不响应中断acquireInterruptibly响应中断tryAcquireNanos带超时和中断响应public final void acquireInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); if (!tryAcquire(arg)) doAcquireInterruptibly(arg); }9.2 自定义同步器示例限流器基于AQS实现一个简单的令牌桶限流器public class RateLimiter { private final Sync sync; private final Timer timer; private static final class Sync extends AbstractQueuedSynchronizer { private final int maxPermits; private long nextFreeTicket; Sync(int permits) { maxPermits permits; setState(permits); nextFreeTicket System.nanoTime(); } Override protected int tryAcquireShared(int acquires) { for (;;) { long now System.nanoTime(); long next nextFreeTicket; long wait Math.max(next - now, 0); if (wait 0) { return -1; } int available getState(); int remaining available - acquires; if (remaining 0 || compareAndSetState(available, remaining)) { nextFreeTicket now 1_000_000_000L / maxPermits; return remaining; } } } Override protected boolean tryReleaseShared(int releases) { for (;;) { int current getState(); int next current releases; if (next maxPermits) { next maxPermits; } if (compareAndSetState(current, next)) { return true; } } } } public RateLimiter(int permits) { this.sync new Sync(permits); this.timer new Timer(true); } public void acquire() throws InterruptedException { sync.acquireSharedInterruptibly(1); } public void release() { sync.releaseShared(1); } }这个实现展示了令牌桶算法原理精确的时间控制AQS共享模式的高级用法10. AQS与现代并发编程随着Java并发API的发展AQS仍然是许多高级并发工具的基础StampedLock结合了AQS和乐观读Phaser更灵活的同步屏障CompletableFuture部分实现依赖AQS理解AQS有助于我们更高效地使用标准库并发工具在必要时实现自定义同步器诊断复杂的并发问题在实际项目中我经常遇到需要基于AQS实现特殊同步需求的场景。比如最近我们实现了一个分布式任务调度系统其中的本地任务队列就使用了自定义的AQS同步器来协调多个工作线程。通过合理设计tryAcquire/tryRelease逻辑我们实现了比标准锁更高效的并发控制。
1. 理解AQS的本质第一次接触AQS时我盯着这个缩写看了半天——AbstractQueuedSynchronizer抽象队列同步器。名字听起来就很唬人但当我真正理解它的设计哲学后才发现这简直是Java并发编程中的瑞士军刀。AQS本质上是一个构建锁和同步器的框架它用了一个int成员变量表示同步状态通过内置的FIFO队列来完成资源获取线程的排队工作。重要提示AQS不是直接给开发者用的工具类而是给那些需要实现自己同步原语的大神们准备的底层框架。我们日常用的ReentrantLock、CountDownLatch等其实都是基于AQS实现的。AQS最精妙的地方在于它采用了模板方法模式。它定义了acquire和release这样的模板方法但把具体的资源获取和释放逻辑留给子类实现。这种设计让AQS既能保持核心排队机制的稳定性又能灵活支持各种同步场景。2. AQS的核心机制解析2.1 同步状态与CAS操作AQS内部维护了一个volatile int类型的state变量这个变量就是同步状态的核心。不同的同步器对state的解释不同在ReentrantLock中state表示锁的持有计数在Semaphore中state表示可用许可数在CountDownLatch中state表示计数器值所有对state的修改都通过CASCompare-And-Swap操作完成这是AQS能够实现无锁并发的基础。在JDK中这个操作通过Unsafe类实现protected final boolean compareAndSetState(int expect, int update) { return unsafe.compareAndSwapInt(this, stateOffset, expect, update); }2.2 等待队列的实现当线程获取资源失败时AQS会将线程封装成一个Node节点加入CLH队列一种改进的CLH锁队列。这个队列是AQS能够实现公平性的关键static final class Node { volatile int waitStatus; volatile Node prev; volatile Node next; volatile Thread thread; Node nextWaiter; // ... }每个Node节点保存了线程引用、前后节点指针以及等待状态。队列的入队和出队操作同样通过CAS保证线程安全。2.3 独占与共享模式AQS支持两种同步模式独占模式Exclusive同一时刻只有一个线程能获取资源如ReentrantLock共享模式Shared多个线程可以同时获取资源如Semaphore这两种模式通过不同的acquire/release方法实现acquire/release独占模式acquireShared/releaseShared共享模式3. 基于AQS实现自定义同步器3.1 实现一个简单的互斥锁让我们通过实现一个简单的不可重入互斥锁来理解AQS的使用public class MyMutex { private static class Sync extends AbstractQueuedSynchronizer { Override protected boolean tryAcquire(int arg) { return compareAndSetState(0, 1); } Override protected boolean tryRelease(int arg) { setState(0); return true; } Override protected boolean isHeldExclusively() { return getState() 1; } } private final Sync sync new Sync(); public void lock() { sync.acquire(1); } public void unlock() { sync.release(1); } }这个实现虽然简单但包含了AQS使用的核心要点继承AQS并实现tryAcquire/tryRelease方法使用state表示锁状态0表示未锁定1表示锁定通过acquire/release方法提供外部接口3.2 实现一个简单的CountDownLatch再来看看如何实现一个简单的CountDownLatchpublic class MyCountDownLatch { private static class Sync extends AbstractQueuedSynchronizer { Sync(int count) { setState(count); } Override protected int tryAcquireShared(int acquires) { return (getState() 0) ? 1 : -1; } Override protected boolean tryReleaseShared(int releases) { for (;;) { int c getState(); if (c 0) return false; int nextc c - 1; if (compareAndSetState(c, nextc)) return nextc 0; } } } private final Sync sync; public MyCountDownLatch(int count) { this.sync new Sync(count); } public void await() throws InterruptedException { sync.acquireSharedInterruptibly(1); } public void countDown() { sync.releaseShared(1); } }这个实现展示了AQS共享模式的使用关键点在于tryAcquireShared返回负值表示获取失败tryReleaseShared通过CAS循环减少计数器当计数器归零时所有等待线程会被唤醒4. AQS在JDK中的应用实例4.1 ReentrantLock的实现ReentrantLock是AQS最经典的用法之一。它的内部类Sync继承自AQS并实现了可重入特性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) // overflow throw new Error(Maximum lock count exceeded); setState(nextc); return true; } return false; }这段代码展示了可重入锁的关键逻辑如果锁未被持有state0尝试获取如果锁已被当前线程持有增加重入计数释放时相应减少计数归零时完全释放4.2 Semaphore的实现Semaphore使用AQS的共享模式来控制资源访问protected int tryAcquireShared(int acquires) { for (;;) { int available getState(); int remaining available - acquires; if (remaining 0 || compareAndSetState(available, remaining)) return remaining; } }Semaphore的tryAcquireShared实现展示了计算剩余许可数如果不足直接返回负值否则尝试CAS更新状态5. AQS性能优化技巧5.1 减少CAS竞争在高并发场景下AQS的CAS操作可能成为瓶颈。我们可以通过以下方式优化使用自旋等待在入队前短暂自旋可能避免队列操作减少state修改频率比如在读写锁中使用位运算分离读写状态使用ThreadLocal缓存对于可重入锁可以缓存部分状态5.2 队列优化AQS的CLH队列可以针对特定场景优化对于短时任务可以设置自旋超时对于读多写少的场景可以偏向读锁使用条件队列时注意避免虚假唤醒6. AQS常见问题排查6.1 死锁问题虽然AQS本身不会导致死锁但使用不当仍然可能发生// 错误示例嵌套锁可能导致死锁 public void transfer(MyMutex lock1, MyMutex lock2) { lock1.lock(); lock2.lock(); try { // 转账操作 } finally { lock2.unlock(); lock1.unlock(); } }解决方案按固定顺序获取锁使用tryLock设置超时使用更高级别的并发工具6.2 性能问题AQS在高竞争场景下可能出现性能问题表现为CAS操作失败率高队列过长导致上下文切换频繁解决方法减小锁粒度使用读写锁分离读写操作考虑无锁算法替代方案7. AQS最佳实践7.1 正确实现tryAcquire实现tryAcquire时需要注意必须保证线程安全考虑可重入性需求正确处理中断protected boolean tryAcquire(int arg) { Thread current Thread.currentThread(); int c getState(); if (c 0) { if (!hasQueuedPredecessors() // 公平性检查 compareAndSetState(0, arg)) { setExclusiveOwnerThread(current); return true; } } else if (current getExclusiveOwnerThread()) { int nextc c arg; if (nextc 0) throw new Error(Maximum lock count exceeded); setState(nextc); return true; } return false; }7.2 合理使用条件队列AQS的条件队列ConditionObject提供了更灵活的等待/通知机制public class BoundedBuffer { private final Lock lock new ReentrantLock(); private final Condition notFull lock.newCondition(); private final Condition notEmpty lock.newCondition(); public void put(Object x) throws InterruptedException { lock.lock(); try { while (count items.length) notFull.await(); items[putPtr] x; if (putPtr items.length) putPtr 0; count; notEmpty.signal(); } finally { lock.unlock(); } } // 类似的take方法... }使用条件队列时要注意总是在循环中检查条件先获取锁再调用await/signal使用多个Condition分离不同的等待条件8. AQS源码深度解析8.1 acquireQueued方法剖析这是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)) { setHead(node); p.next null; // help GC failed false; return interrupted; } if (shouldParkAfterFailedAcquire(p, node) parkAndCheckInterrupt()) interrupted true; } } finally { if (failed) cancelAcquire(node); } }这个方法展示了只有前驱是头节点时才尝试获取公平性获取失败后检查是否需要阻塞处理中断标志最终处理失败情况8.2 取消机制分析当线程在等待过程中被中断或超时需要取消获取操作private void cancelAcquire(Node node) { if (node null) return; node.thread null; // 跳过已取消的前驱 Node pred node.prev; while (pred.waitStatus 0) node.prev pred pred.prev; Node predNext pred.next; node.waitStatus Node.CANCELLED; // 如果是尾节点直接移除 if (node tail compareAndSetTail(node, pred)) { compareAndSetNext(pred, predNext, null); } else { // 否则唤醒后继节点 int ws; if (pred ! head ((ws pred.waitStatus) Node.SIGNAL || (ws 0 compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) pred.thread ! null) { Node next node.next; if (next ! null next.waitStatus 0) compareAndSetNext(pred, predNext, next); } else { unparkSuccessor(node); } node.next node; // help GC } }取消操作需要考虑清理线程引用跳过已取消的前驱处理不同位置的节点必要时唤醒后继9. AQS扩展与变种9.1 响应中断的获取操作AQS提供了多种获取方法应对不同需求acquire不响应中断acquireInterruptibly响应中断tryAcquireNanos带超时和中断响应public final void acquireInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); if (!tryAcquire(arg)) doAcquireInterruptibly(arg); }9.2 自定义同步器示例限流器基于AQS实现一个简单的令牌桶限流器public class RateLimiter { private final Sync sync; private final Timer timer; private static final class Sync extends AbstractQueuedSynchronizer { private final int maxPermits; private long nextFreeTicket; Sync(int permits) { maxPermits permits; setState(permits); nextFreeTicket System.nanoTime(); } Override protected int tryAcquireShared(int acquires) { for (;;) { long now System.nanoTime(); long next nextFreeTicket; long wait Math.max(next - now, 0); if (wait 0) { return -1; } int available getState(); int remaining available - acquires; if (remaining 0 || compareAndSetState(available, remaining)) { nextFreeTicket now 1_000_000_000L / maxPermits; return remaining; } } } Override protected boolean tryReleaseShared(int releases) { for (;;) { int current getState(); int next current releases; if (next maxPermits) { next maxPermits; } if (compareAndSetState(current, next)) { return true; } } } } public RateLimiter(int permits) { this.sync new Sync(permits); this.timer new Timer(true); } public void acquire() throws InterruptedException { sync.acquireSharedInterruptibly(1); } public void release() { sync.releaseShared(1); } }这个实现展示了令牌桶算法原理精确的时间控制AQS共享模式的高级用法10. AQS与现代并发编程随着Java并发API的发展AQS仍然是许多高级并发工具的基础StampedLock结合了AQS和乐观读Phaser更灵活的同步屏障CompletableFuture部分实现依赖AQS理解AQS有助于我们更高效地使用标准库并发工具在必要时实现自定义同步器诊断复杂的并发问题在实际项目中我经常遇到需要基于AQS实现特殊同步需求的场景。比如最近我们实现了一个分布式任务调度系统其中的本地任务队列就使用了自定义的AQS同步器来协调多个工作线程。通过合理设计tryAcquire/tryRelease逻辑我们实现了比标准锁更高效的并发控制。
