【深入解析AQS】从设计模式到ReentrantLock实现再到自定义锁

技术分享 9个月前 (05-01) 0 999+

关注

深入解析AQS：设计模式、ReentrantLock实现与自定义锁开发

一、模板方法模式：AQS的架构基石

1.1 模式核心思想

模板方法模式通过固定算法骨架+可变实现细节的设计，实现了代码复用与扩展性的平衡。AQS采用这种模式，将同步器的核心流程（如线程排队、阻塞唤醒）固化在父类，仅将资源获取/释放的逻辑通过抽象方法交给子类实现。

设计优势：

保证正确性：关键同步流程不可修改
提高复用：通用逻辑只需实现一次
便于扩展：子类只需关注业务逻辑

1.2 完整代码示例

// 模板类定义 abstract class BeverageMaker {     // 模板方法（final防止重写）     public final void makeBeverage() {         boilWater();         brew();         pourInCup();         if (customerWantsCondiments()) {             addCondiments();         }     }          // 具体方法（通用步骤）     private void boilWater() {         System.out.println("煮沸水");     }          private void pourInCup() {         System.out.println("倒入杯中");     }          // 抽象方法（子类必须实现）     protected abstract void brew();     protected abstract void addCondiments();          // 钩子方法（子类可选覆盖）     protected boolean customerWantsCondiments() {         return true;     } }  // 具体实现类 class TeaMaker extends BeverageMaker {     @Override     protected void brew() {         System.out.println("浸泡茶叶5分钟");     }      @Override     protected void addCondiments() {         System.out.println("加入柠檬片");     }          @Override     protected boolean customerWantsCondiments() {         return false; // 茶不需要调料     } }

关键点说明：

makeBeverage()定义了不可变的制作流程
brew()和addCondiments()是可变部分，由子类实现
customerWantsCondiments()是钩子方法，提供策略扩展点

1.3 AQS中的模板模式实现

AQS将同步操作抽象为模板方法：

public abstract class AbstractQueuedSynchronizer {     // 获取资源的模板方法     public final void acquire(int arg) {         if (!tryAcquire(arg) &&  // 尝试获取（子类实现）             acquireQueued(addWaiter(Node.EXCLUSIVE), arg))             selfInterrupt();     }          // 由子类实现的获取逻辑     protected boolean tryAcquire(int arg) {         throw new UnsupportedOperationException();     }          // 已实现的通用方法     private Node addWaiter(Node mode) {         // 实现节点入队逻辑...     }          final boolean acquireQueued(Node node, int arg) {         // 实现队列中获取资源的逻辑...     } }

设计精妙之处：

acquire()封装了完整的获取资源流程
子类只需关注tryAcquire的核心逻辑
线程排队、阻塞唤醒等复杂操作由AQS统一处理

二、ReentrantLock与AQS的深度整合

2.1 整体架构设计

ReentrantLock通过内部类继承AQS，实现锁的核心功能：

classDiagram class ReentrantLock { -Sync sync +lock() +unlock() } class Sync { <<abstract>> +nonfairTryAcquire() +tryRelease() } class NonfairSync { +lock() +tryAcquire() } class FairSync { +tryAcquire() } ReentrantLock --> Sync Sync <|-- NonfairSync Sync <|-- FairSync Sync --> AQS

2.2 关键源码解析

非公平锁实现：

static final class NonfairSync extends Sync {     // 非公平获取锁     final void lock() {         if (compareAndSetState(0, 1))  // 先尝试快速获取             setExclusiveOwnerThread(Thread.currentThread());         else             acquire(1);  // 进入AQS排队流程     }          protected final boolean tryAcquire(int acquires) {         return nonfairTryAcquire(acquires);     } }  // Sync中的通用非公平尝试 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) throw new Error("Maximum lock count exceeded");         setState(nextc);         return true;     }     return false; }

实现特点：

快速路径：先直接尝试CAS获取锁，避免排队开销
重入支持：通过检查当前线程和状态计数实现
非公平性：新请求线程可能插队获取锁

公平锁实现差异：

static final class FairSync extends Sync {     protected final boolean tryAcquire(int acquires) {         final Thread current = Thread.currentThread();         int c = getState();         if (c == 0) {             if (!hasQueuedPredecessors() &&  // 关键区别：检查队列                 compareAndSetState(0, acquires)) {                 setExclusiveOwnerThread(current);                 return true;             }         }         // 重入逻辑与非公平锁相同...     } }

公平性保证：

hasQueuedPredecessors()检查是否有更早等待的线程
严格按照CLH队列顺序获取锁
吞吐量通常低于非公平锁，但避免线程饥饿

2.3 状态管理机制

AQS使用volatile int state字段记录同步状态，ReentrantLock中表示：

0：锁未被任何线程持有
>0：锁被持有，数值表示重入次数

配套原子操作方法：

protected final int getState() {     return state; }  protected final void setState(int newState) {     state = newState; }  protected final boolean compareAndSetState(int expect, int update) {     return unsafe.compareAndSwapInt(this, stateOffset, expect, update); }

三、基于AQS实现自定义锁

3.1 完整自定义锁实现

import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.AbstractQueuedSynchronizer; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock;  /**  * 基于AQS的简单互斥锁（不可重入）  */ public class SimpleMutexLock implements Lock {     private final Sync sync = new Sync();      // 内部同步器     private static class Sync extends AbstractQueuedSynchronizer {         // 尝试获取锁         @Override         protected boolean tryAcquire(int acquires) {             if (acquires != 1) throw new IllegalArgumentException();             if (compareAndSetState(0, 1)) {  // CAS保证原子性                 setExclusiveOwnerThread(Thread.currentThread());                 return true;             }             return false;         }          // 尝试释放锁         @Override         protected boolean tryRelease(int releases) {             if (releases != 1) throw new IllegalArgumentException();             if (getState() == 0) throw new IllegalMonitorStateException();             setExclusiveOwnerThread(null);             setState(0);  // 不需要CAS，只有持有线程能调用             return true;         }          // 是否被当前线程独占         @Override         protected boolean isHeldExclusively() {             return getState() == 1;         }          // 创建条件变量         Condition newCondition() {             return new ConditionObject();         }     }      // ========== Lock接口实现 ==========     @Override     public void lock() {         sync.acquire(1);     }      @Override     public void lockInterruptibly() throws InterruptedException {         sync.acquireInterruptibly(1);     }      @Override     public boolean tryLock() {         return sync.tryAcquire(1);     }      @Override     public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {         return sync.tryAcquireNanos(1, unit.toNanos(time));     }      @Override     public void unlock() {         sync.release(1);     }      @Override     public Condition newCondition() {         return sync.newCondition();     }      // ========== 扩展方法 ==========     public boolean isLocked() {         return sync.isHeldExclusively();     }          public boolean hasQueuedThreads() {         return sync.hasQueuedThreads();     } }

实现要点：

状态定义：0表示未锁定，1表示锁定
不可重入：不检查当前线程是否已持有锁
条件变量：直接复用AQS的ConditionObject
线程安全：CAS操作保证tryAcquire的原子性

3.2 可重入锁改造

修改Sync内部类即可实现可重入：

private static class Sync extends AbstractQueuedSynchronizer {     @Override     protected boolean tryAcquire(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) throw new Error("Maximum lock count exceeded");             setState(nextc);             return true;         }         return false;     }      @Override     protected 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;     } }

关键修改点：

增加当前线程检查实现重入
通过状态计数记录重入次数
只有计数归零时才完全释放锁

3.3 完整测试用例

public class SimpleMutexLockTest {     public static void main(String[] args) throws InterruptedException {         // 基础功能测试         testBasicOperation();                  // 并发安全测试         testConcurrentAccess();                  // 不可重入性测试         testNonReentrant();     }      private static void testBasicOperation() {         SimpleMutexLock lock = new SimpleMutexLock();         assert !lock.isLocked() : "初始状态应该未锁定";                  lock.lock();         try {             assert lock.isLocked() : "锁定后状态应该为true";             assert !lock.tryLock() : "不可重入锁应获取失败";         } finally {             lock.unlock();         }     }      private static void testConcurrentAccess() throws InterruptedException {         final int THREADS = 5;         final SimpleMutexLock lock = new SimpleMutexLock();         final AtomicInteger counter = new AtomicInteger();         final CountDownLatch startLatch = new CountDownLatch(1);         final CountDownLatch endLatch = new CountDownLatch(THREADS);          ExecutorService executor = Executors.newFixedThreadPool(THREADS);         for (int i = 0; i < THREADS; i++) {             executor.execute(() -> {                 try {                     startLatch.await();                     lock.lock();                     try {                         counter.incrementAndGet();                         Thread.sleep(100);  // 模拟操作                     } finally {                         lock.unlock();                         endLatch.countDown();                     }                 } catch (InterruptedException e) {                     Thread.currentThread().interrupt();                 }             });         }                  startLatch.countDown();         endLatch.await();         assert counter.get() == THREADS : "所有线程应该都执行了";         executor.shutdown();     }      private static void testNonReentrant() {         SimpleMutexLock lock = new SimpleMutexLock();         lock.lock();         try {             // 会阻塞在这里，因为是不可重入锁             boolean acquired = lock.tryLock();              assert !acquired : "不可重入锁不应再次获取成功";         } finally {             lock.unlock();         }     } }