1. 概述

全称是 AbstractQueuedSynchronizer，是阻塞式锁和相关的同步器工具的框架，许多同步类实现都依赖于该同步器。

AQS 用状态属性来表示资源的状态（分独占模式和共享模式），子类需要定义如何维护这个状态，控制如何获取锁和释放锁

• 独占模式是只有一个线程能够访问资源，如 ReentrantLock
• 共享模式允许多个线程访问资源，如 Semaphore，ReentrantReadWriteLock 是组合式

AQS 核心思想：

• 如果被请求的共享资源空闲，则将当前请求资源的线程设置为有效的工作线程，并将共享资源设置锁定状态
• 请求的共享资源被占用，AQS 用队列实现线程阻塞等待以及被唤醒时锁分配的机制，将暂时获取不到锁的线程加入到队列中

CLH 是一种基于单向链表的高性能、公平的自旋锁，AQS 是将每条请求共享资源的线程封装成一个 CLH 锁队列的一个结点（Node）来实现锁的分配

1.1 特点

特点：

• state 属性：独占 / 共享模式
  • 用 state 属性来表示资源的状态（分独占模式和共享模式），子类需要定义如何维护这个状态，控制如何获取锁和释放锁
    • getState - 获取 state 状态
    • setState - 设置 state 状态
    • compareAndSetState - CAS 机制设置 state 状态
    • 独占模式是只有一个线程能够访问资源，而共享模式可以允许多个线程访问资源
• 等待队列：提供了基于 FIFO 的等待队列，类似于 Monitor 的 EntryList
• 条件变量：条件变量来实现等待、唤醒机制，支持多个条件变量，类似于 Monitor 的 WaitSet

1.2 实现 AQS

使用 AQS 的方式就是实现一个子类，并继承这个 AQS 父类。子类主要实现这样一些方法（默认抛出 UnsupportedOperationException）：

• tryAcquire
• tryRelease
• tryAcquireShared
• tryReleaseShared
• isHeldExclusivel

获取锁的姿势：

// 如果获取锁失败
if (!tryAcquire(arg)) {
     // 入队, 可以选择阻塞当前线程 park unpark
}

释放锁的姿势：

// 如果释放锁成功
if (tryRelease(arg)) {
   // 让阻塞线程恢复运行
}

2. 实现不可重入锁

不可重入锁是指自己上的锁自己也进不去。

2.1 自定义同步器

实现的锁的大部分功能都是由这个同步器类来实现的，它继承了 AQS：

final class MySync extends AbstractQueuedSynchronizer {
    
    @Override
    protected boolean tryAcquire(int acquires) {
        if (acquires == 1){
            if (compareAndSetState(0, 1)) {
                // 设置 owner 为当前线程
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
        }
        return false;
    }
    
    @Override
    protected boolean tryRelease(int acquires) {
        if(acquires == 1) {
            if(getState() == 0) {
                throw new IllegalMonitorStateException();
            }
            setExclusiveOwnerThread(null);
            setState(0);  // state 是 volatile 的，这里 set 会有写屏障，因此把这行代码放后面防止指令重排序
            return true;
        }
        return false;
    }
    
    protected Condition newCondition() {
        return new ConditionObject();
    }
    
    // 是否持有独占锁
    @Override
    protected boolean isHeldExclusively() {
        return getState() == 1;
    }
    
}

AQS 已经把大部分方法都写好了，我们只需要实现那么几个函数就可以了。

2.2 自定义锁

有了自定义同步器，很容易复用 AQS ，实现一个功能完备的自定义锁：

class MyLock implements Lock {
    
    static MySync sync = new MySync();
    
    @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();
    }
    
}

测试一下：

MyLock lock = new MyLock();

new Thread(() -> {
    lock.lock();
    try {
        log.debug("locking...");
        sleep(1);
    } finally {
        log.debug("unlocking...");
        lock.unlock();
    }
},"t1").start();

new Thread(() -> {
    lock.lock();
    try {
        log.debug("locking...");
    } finally {
        log.debug("unlocking...");
        lock.unlock();
    }
},"t2").start();

输出：

22:29:28.727 c.TestAqs [t1] - locking... 
22:29:29.732 c.TestAqs [t1] - unlocking... 
22:29:29.732 c.TestAqs [t2] - locking... 
22:29:29.732 c.TestAqs [t2] - unlocking...

不可重入测试。如果改为下面代码，会发现自己也会被挡住（只会打印一次 locking）：

lock.lock();
log.debug("locking...");
lock.lock();
log.debug("locking...");

3. 心得

3.1 起源

早期程序员会自己通过一种同步器去实现另一种相近的同步器，例如用可重入锁去实现信号量，或反之。这显然不够优雅，于是在 JSR166（java 规范提案）中创建了 AQS，提供了这种通用的同步器机制。

3.2 目标

AQS 要实现的功能目标：

• 阻塞版本获取锁 acquire 和非阻塞的版本尝试获取锁 tryAcquire
• 获取锁超时机制
• 通过打断取消机制
• 独占机制及共享机制
• 条件不满足时的等待机制

要实现的性能目标：

Instead, the primary performance goal here is scalability: to predictably maintain effiffifficiency even, or especially, when synchronizers are contended.

3.3 设计

AQS 的基本思想其实很简单。

获取锁的逻辑：

while (state 状态不允许获取) {
    if (队列中还没有此线程) {
        入队并阻塞
    }
}
当前线程出队

释放锁的逻辑：

if (state 状态允许了) {
    恢复阻塞的线程(s) 
}

要点：

• 原子维护 state 状态
• 阻塞及恢复线程
• 维护队列

3.3.1 state 设计

• state 使用 volatile 配合 cas 保证其修改时的原子性
• state 使用了 32bit int 来维护同步状态，因为当时使用 long 在很多平台下测试的结果并不理想

3.3.2 阻塞恢复设计

• 早期的控制线程暂停和恢复的 api 有 suspend 和 resume，但它们是不可用的，因为如果先调用的 resume 那么 suspend 将感知不到
• 解决方法是使用 park & unpark 来实现线程的暂停和恢复，具体原理在之前讲过了，先 unpark 再 park 也没 问题
• park & unpark 是针对线程的，而不是针对同步器的，因此控制粒度更为精细
• park 线程还可以通过 interrupt 打断

3.3.3 队列设计

• 使用了 FIFO 先入先出队列，并不支持优先级队列
• 设计时借鉴了 CLH 队列，它是一种单向无锁队列

队列中有 head 和 tail 两个指针节点，都用 volatile 修饰配合 cas 使用，每个节点有 state 维护节点状态

入队伪代码，只需要考虑 tail 赋值的原子性：

do {
    // 原来的 tail
    Node prev = tail;
    // 用 cas 在原来 tail 的基础上改为 node
} while (tail.compareAndSet(prev, node))

出队伪代码：

// prev 是上一个节点
while((Node prev=node.prev).state != 唤醒状态) {
}
// 设置头节点
head = node;

CLH 好处：

• 无锁，使用自旋
• 快速，无阻塞

AQS 在一些方面改进了 CLH：

private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        // 队列中还没有元素 tail 为 null
        if (t == null) {
            // 将 head 从 null -> dummy
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            // 将 node 的 prev 设置为原来的 tail
            node.prev = t;
            // 将 tail 从原来的 tail 设置为 node
            if (compareAndSetTail(t, node)) {
                // 原来 tail 的 next 设置为 node
                t.next = node;
                return t;
            }
        }
    }
}