【JVM深度解析】第26篇：CAS、AQS与并发工具类原理

摘要CASCompare-And-Swap和 AQSAbstractQueuedSynchronizer是 Java 并发包的基石。CAS 通过硬件支持的原子指令实现无锁并发AQS 通过模板模式封装了线程等待和唤醒的通用逻辑。本文深入解析 CAS 的底层实现Unsafe 类 CPU 原子指令、AQS 的核心原理CLH 队列 state 变量、以及 ReentrantLock、Semaphore、CountDownLatch、CyclicBarrier 等常用并发工具的实现细节。掌握这些你才能真正理解 concurrent 包的设计哲学写出高效的无锁代码。一、CAS 深度解析1.1 CAS 原理CAS 操作 ┌──────────────────────────────────────────────────────────────────┐ │ │ │ CAS(V, Expected, New) { │ │ if (V Expected) { │ │ V New; │ │ return true; │ │ } │ │ return false; │ │ } │ │ │ │ 特点 │ │ - 单个原子操作 │ │ - 硬件支持CPU 指令级 │ │ - 不阻塞线程失败重试 │ │ │ └──────────────────────────────────────────────────────────────────┘1.2 Unsafe 类的 CAS 操作// Java 中 CAS 的使用publicclassAtomicInteger{privatevolatileintvalue;// sun.misc.Unsafe - 直接操作内存的类privatestaticfinalUnsafeunsafeUnsafe.getUnsafe();privatestaticfinallongvalueOffset;static{try{valueOffsetunsafe.objectFieldOffset(AtomicInteger.class.getDeclaredField(value));}catch(Exceptione){thrownewError(e);}}// 原子递增publicfinalintincrementAndGet(){returnunsafe.getAndAddInt(this,valueOffset,1)1;}// Unsafe 的 getAndAddInt 实现// public final int getAndAddInt(Object o, long offset, int delta) {// int v;// do {// v unsafe.getIntVolatile(o, offset); // 读取当前值// } while (!unsafe.compareAndSwapInt(o, offset, v, v delta)); // CAS// return v;// }}1.3 CAS 的 ABA 问题ABA 问题 ┌──────────────────────────────────────────────────────────────────┐ │ │ │ 线程 A 读取 V 1 │ │ 线程 B: CAS(V, 1, 2) 成功V 2 │ │ 线程 B: CAS(V, 2, 1) 成功V 1 ← 变回来了 │ │ 线程 A: CAS(V, 1, 3) 成功 ← A 以为没变过 │ │ │ │ 后果某些场景下会导致错误结果 │ │ │ └──────────────────────────────────────────────────────────────────┘// 解决方案AtomicStampedReference带版本号publicclassABASolution{privatestaticAtomicStampedReferenceIntegerrefnewAtomicStampedReference(1,0);publicstaticvoidmain(String[]args){intstampref.getStamp();intexpectedref.getReference();// CAS 加上版本号检查ref.compareAndSet(expected,2,stamp,stamp1);ref.compareAndSet(2,1,stamp1,stamp2);// 现在线程 A 会失败版本号不匹配ref.compareAndSet(1,3,stamp,stamp1);// false}}二、AQS 深度解析2.1 AQS 核心结构┌──────────────────────────────────────────────────────────────────┐ │ AQS 核心结构 │ ├──────────────────────────────────────────────────────────────────┤ │ │ │ ┌──────────────────────────────────────────────────────────┐ │ │ │ state 状态变量 │ │ │ │ │ │ │ │ - ReentrantLock: 持有锁的次数 │ │ │ │ - Semaphore: 可用许可数 │ │ │ │ - CountDownLatch: 倒计时计数 │ │ │ │ - CyclicBarrier: 等待线程数 │ │ │ └──────────────────────────────────────────────────────────┘ │ │ │ │ ┌──────────────────────────────────────────────────────────┐ │ │ │ CLH 队列双向链表 │ │ │ │ │ │ │ │ head ──→ node ──→ node ──→ node ──→ tail │ │ │ │ (持有者) │ │ │ │ │ │ │ │ Node 属性 │ │ │ │ - prev / next: 双向链表指针 │ │ │ │ - thread: 等待的线程 │ │ │ │ - waitStatus: 等待状态 │ │ │ │ - SHARED / EXCLUSIVE: 共享/独占模式 │ │ │ └──────────────────────────────────────────────────────────┘ │ │ │ └──────────────────────────────────────────────────────────────────┘2.2 ReentrantLock 实现// ReentrantLock 基于 AQS 的实现publicclassReentrantLockextendsAbstractQueuedSynchronizer{// 1. 非公平锁 tryAcquireprotectedbooleantryAcquire(intacquires){finalThreadcurrentThread.currentThread();intcgetState();if(c0){// CAS 尝试获取锁if(compareAndSetState(0,acquires)){setExclusiveOwnerThread(current);returntrue;}}elseif(currentgetExclusiveOwnerThread()){// 重入增加 statesetState(cacquires);returntrue;}returnfalse;}// 2. tryReleaseprotectedbooleantryRelease(intreleases){intcgetState()-releases;if(Thread.currentThread()!getExclusiveOwnerThread())thrownewIllegalMonitorStateException();booleanfreefalse;if(c0){freetrue;setExclusiveOwnerThread(null);}setState(c);returnfree;}// 3. lock() 方法publicvoidlock(){sync.lock();// 委托给 Sync}}2.3 CountDownLatch 实现// CountDownLatch 基于 AQS 的实现publicclassCountDownLatchextendsAbstractQueuedSynchronizer{privatestaticfinalclassSyncextendsAbstractQueuedSynchronizer{Sync(intcount){setState(count);// state 计数}// 共享模式获取等待计数归零protectedinttryAcquireShared(intignored){returngetState()0?1:-1;}// 共享模式释放计数 - 1protectedbooleantryReleaseShared(intdecrements){for(;;){intcgetState();if(c0)returnfalse;intnextc-1;if(compareAndSetState(c,next))returnnext0;}}}privatefinalSyncsync;// await() - 等待计数归零publicvoidawait()throwsInterruptedException{sync.acquireSharedInterruptibly(1);}// countDown() - 计数 - 1publicvoidcountDown(){sync.releaseShared(1);}}三、常用并发工具对比┌──────────────────────────────────────────────────────────────────┐ │ Java 并发工具对比 │ ├──────────────────────────────────────────────────────────────────┤ │ │ │ ReentrantLock vs synchronized: │ │ ┌────────────────────────────────────────────────────────────┐ │ │ │ 特性 │ ReentrantLock │ synchronized │ │ │ │ 可中断 │ 支持 │ 不支持 │ │ │ │ 超时 │ 支持 │ 不支持 │ │ │ │ 非公平锁 │ 支持 │ 不支持非公平 │ │ │ │ 多条件队列 │ 支持 │ 不支持 │ │ │ │ 性能 │ JDK 6 相近 │ JDK 6 相近 │ │ │ └────────────────────────────────────────────────────────────┘ │ │ │ │ CountDownLatch vs CyclicBarrier: │ │ ┌────────────────────────────────────────────────────────────┐ │ │ │ 特性 │ CountDownLatch │ CyclicBarrier │ │ │ │ 用途 │ 一次性等待 │ 可循环使用 │ │ │ │ 计数器 │ 只减不复位 │ 归零后重置 │ │ │ │ 复用 │ 不可 │ 可 │ │ │ │ 等待线程状态 │ 继续执行 │ 一起继续执行 │ │ │ └────────────────────────────────────────────────────────────┘ │ │ │ └──────────────────────────────────────────────────────────────────┘四、总结CAS 是硬件级的原子操作Unsafe 类提供了直接的 CAS 调用但存在 ABA 问题可用 AtomicStampedReference 解决。AQS 通过 state 变量和 CLH 队列封装了线程等待/唤醒的通用逻辑ReentrantLock、Semaphore、CountDownLatch、CyclicBarrier 等工具都基于 AQS 构建。理解这些底层原理才能真正用好并发包。系列导航上一篇【JVM深度解析】第25篇volatile与synchronized深度原理下一篇【JVM深度解析】第27篇并发编程实战案例与陷阱系列目录JVM深度解析参考资料JUC - AbstractQueuedSynchronizerDoug Lea - The java.util.concurrent Synchronizer FrameworkJava Concurrency in Practice