Java Aqs原理与使用

AQS（AbstractQueuedSynchronizer，抽象队列同步器）是 JUC（java.util.concurrent）包的核心基石。它提供了一套基于 FIFO 等待队列的同步框架，用于构建锁和其他同步组件。

AQS 的设计目标： 将同步器的通用逻辑（队列管理、线程阻塞/唤醒）封装在抽象类中，让子类只需实现少量状态控制方法即可构建完整的同步器。

核心原理

1. 状态管理

   AQS 通过一个 volatile int 变量 state 表示同步状态，提供三个方法操作：

   • getState() — 获取当前状态
   • setState(int) — 设置状态值
   • compareAndSetState(int expect, int update) — CAS 原子更新

2. 模板方法模式

   AQS 采用模板方法模式，对外暴露的 “lock / unlock” 类方法（如 acquire() 、 release() ）已经定义好完整流程，内部留出 5 个钩子方法供子类覆写：

   方法	用途
   tryAcquire(int)	独占模式获取
   tryRelease(int)	独占模式释放
   tryAcquireShared(int)	共享模式获取
   tryReleaseShared(int)	共享模式释放
   isHeldExclusively()	是否被当前线程独占

   子类只需实现这些方法，无需关心线程排队、阻塞、唤醒等复杂细节。

3. CLH 队列

   AQS 内部维护一个 CLH 变体（Craig, Landin, and Hagersten）锁队列，结构如下：

           +------+        +------+        +------+
   head -> | Node | <----> | Node | <----> | Node | <- tail
           |thread|        |thread|        |thread|
           |waitSt|        |waitSt|        |waitSt|
           +------+        +------+        +------+
   

   • head 指向哨兵节点（已获取锁的线程）， tail 指向队尾
   • 每个 Node 包含等待线程引用、等待状态（ CANCELLED 、 SIGNAL 、 CONDITION 、 PROPAGATE ）、前驱/后继指针
   • 新节点以 CAS 方式入队，保证线程安全

4. 独占模式 (acquire) 流程

   1. 调用 tryAcquire(arg) 尝试获取锁，成功则直接返回
   2. 失败则调用 addWaiter() 将当前线程封装为独占模式的 Node 加入队尾
   3. 加入队尾后，调用 acquireQueued() 进入自旋循环：
      • 若前驱是 head 且 tryAcquire 成功，将自己设为 head 并返回
      • 否则将前驱的 waitStatus 设为 SIGNAL ，然后 LockSupport.park() 阻塞自己
   4. 被唤醒后回到步骤 3 继续竞争

   独占模式（同一时刻只能一个线程拿到资源）典型实现：ReentrantLock 、synchronized
   

5. 共享模式 (acquireShared) 流程

   与独占模式的核心区别在于：锁释放时会传播唤醒，即一个节点被唤醒后，如果还有剩余资源，会继续唤醒后继节点，实现 “连锁唤醒”。

   共享模式（多个线程可以同时获得资源（允许共享）典型实现： Semaphore 、CountDownLatch 、 ReadLock
   

6. 公平与非公平

   公平锁：严格按队列FIFO执行，新线程必须先排队，不能插队。
   非公平锁：新线程可以直接CAS抢锁，失败才进入队列与排队线程竞争。
   非公平性能更好：线程来了可以直接抢锁，不必先入队再等待唤醒

7. Condition 条件队列

   AQS 内部还维护一个 ConditionObject*，用于实现 *await/signal 机制：

   • 每个 Condition 维护一个单向条件队列
   • await() 将当前线程从同步队列转移到条件队列，释放锁并阻塞
   • signal() 将条件队列头节点移回同步队列，等待重新获取锁

   Condition 解决的不是性能问题，而是“唤醒语义错误问题”，避免无效唤醒。
   

   import java.util.LinkedList;
   import java.util.concurrent.locks.Condition;
   import java.util.concurrent.locks.ReentrantLock;
   
   public class SimpleConditionDemo {
   
       private final LinkedList<Integer> queue = new LinkedList<>();
       private final int MAX = 2;
   
       private final ReentrantLock lock = new ReentrantLock();
       private final Condition notFull = lock.newCondition();
       private final Condition notEmpty = lock.newCondition();
   
       // 生产
       public void put(int v) throws InterruptedException {
           lock.lock();
           try {
               while (queue.size() == MAX) {
                   notFull.await(); // 🔴 满了就等（释放锁 + 进入条件队列）
               }
   
               queue.add(v);
               System.out.println("produce " + v);
   
               notEmpty.signal(); // 🔵 通知消费者
           } finally {
               lock.unlock();
           }
       }
   
       // 消费
       public void take() throws InterruptedException {
           lock.lock();
           try {
               while (queue.isEmpty()) {
                   notEmpty.await(); // 🔴 空了就等
               }
   
               int v = queue.removeFirst();
               System.out.println("consume " + v);
   
               notFull.signal(); // 🔵 通知生产者
           } finally {
               lock.unlock();
           }
       }
   
       public static void main(String[] args) {
           SimpleConditionDemo d = new SimpleConditionDemo();
   
           new Thread(() -> {
               int i = 0;
               while (true) {
                   try { d.put(i++); Thread.sleep(200); }
                   catch (Exception ignored) {}
               }
           }).start();
   
           new Thread(() -> {
               while (true) {
                   try { d.take(); Thread.sleep(500); }
                   catch (Exception ignored) {}
               }
           }).start();
       }
   }
   

常见实现

1. ReentrantLock（可重入锁）

   • 模式： 独占模式

   • 原理： state 记录持有锁的线程和重入次数。 state 0 表示未锁定；>0 表示已锁定，值为重入次数。同一线程可多次 lock() ，每次 state+1 ， unlock() 时 state-1 ，减到 0 时完全释放。

     ReentrantLock lock = new ReentrantLock(); lock.lock(); try { // 临界区代码 } finally { lock.unlock(); }

   • 公平 vs 非公平： 公平锁在 tryAcquire 时先检查队列中是否有等待者；非公平锁直接 CAS 抢占，性能更高但可能饥饿。

2. ReentrantReadWriteLock（读写锁）

   • 模式： 共享 + 独占混合模式

   • 原理： 将 32 位 state 拆分为高 16 位（读锁计数）和低 16 位（写锁计数）。读锁为共享模式，允许多线程并发读；写锁为独占模式，独占写。读-写互斥，写-写互斥。

     ReadWriteLock rwLock = new ReentrantReadWriteLock(); rwLock.readLock().lock(); // 读锁 rwLock.writeLock().lock(); // 写锁

   • 锁降级： 持有写锁时可获取读锁，然后释放写锁，实现从写锁降级到读锁（保证数据可见性）。

3. Semaphore（信号量）

   • 模式： 共享模式

   • 原理： state 表示剩余许可数。 acquire() 时 state-1 ，若 state<0 则阻塞； release() 时 state+1 ，并唤醒等待线程。

     Semaphore semaphore = new Semaphore(3); // 最多 3 个并发 semaphore.acquire(); try { // 受控并发代码 } finally { semaphore.release(); }

4. CountDownLatch（倒计时门闩）

   • 模式： 共享模式

   • 原理： state 为初始计数值。 countDown() 将 state-1 ； await() 在 state 0 之前一直阻塞。当 state 变为 0 时，一次性唤醒所有等待线程。

     CountDownLatch latch = new CountDownLatch(3); // 三个线程各自调用 latch.countDown(); latch.await(); // 阻塞直到 count*0

5. CyclicBarrier（循环栅栏）

   • CyclicBarrier 本身不直接继承 AQS，而是借助 ReentrantLock 和 Condition 实现。

   • 一组线程互相等待，直到全部到达屏障点后同时继续执行，可重复使用。

     import java.util.concurrent.CyclicBarrier;

     public class CyclicBarrierCycleDemo {

     public static void main(String[] args) {
     
         CyclicBarrier barrier = new CyclicBarrier(3,
                 () -> System.out.println(">>> 本轮结束，进入下一轮 <<<"));
     
         Runnable task = () -> {
             try {
                 String name = Thread.currentThread().getName();
     
                 for (int i = 1; i <= 3; i++) {
     
                     System.out.println(name + "：第 " + i + " 轮第一阶段");
     
                     Thread.sleep((long) (Math.random() * 500));
     
                     System.out.println(name + "：到达屏障 " + i);
     
                     barrier.await(); // 🔥 每一轮都会在这里集合
     
                     System.out.println(name + "：第 " + i + " 轮第二阶段完成\n");
                 }
     
             } catch (Exception e) {
                 e.printStackTrace();
             }
         };
     
         new Thread(task, "A").start();
         new Thread(task, "B").start();
         new Thread(task, "C").start();
     }
     

     }

与 synchronized 对比

选择建议