AQS 原理
概述
全称是 AbstractQueuedSynchronizer，是阻塞式锁和相关的同步器工具的框架
特点：
- 用 state 属性来表示资源的状态（分独占模式和共享模式），子类需要定义如何维护这个状态，控制如何获取锁和释放锁
getState - 获取 state 状态
setState - 设置 state 状态
compareAndSetState - cas 机制设置 state 状态
独占模式是只有一个线程能够访问资源，而共享模式可以允许多个线程访问资源
- 提供了基于 FIFO 的等待队列，类似于 Monitor 的 EntryList（Monitor是在C++层面实现的，而 这里的等待队列是在Java层面实现的）
- 条件变量来实现等待、唤醒机制，支持多个条件变量，类似于 Monitor 的 WaitSet
子类主要实现这样一些方法（默认抛出 UnsupportedOperationException）
tryAcquire
tryRelease
tryAcquireShared 方法是一个尝试获取共享资源的方法
tryReleaseShared 方法是一个尝试获取共享资源的方法
isHeldExclusively 方法用于判断当前线程是否独占地持有资源
获取锁的姿势
// 如果获取锁失败 if (!tryAcquire(arg)) { // 入队, 可以选择阻塞当前线程 park unpark（为什么是park 和 unpark在后面介绍源码的时候会展示） }
释放锁的姿势
// 如果释放锁成功 if (tryRelease(arg)) { // 让阻塞线程恢复运行 }
实现不可重入锁
自定义同步器
// 独占锁 同步器类 class MySync extends AbstractQueuedSynchronizer { @Override protected boolean tryAcquire(int arg) { if(compareAndSetState(0, 1)) { // 加上了锁，并设置 owner 为当前线程 和monitor是一致的 setExclusiveOwnerThread(Thread.currentThread()); return true; } return false; } @Override protected boolean tryRelease(int arg) { // 由于state是volatile的，可以防止指令的重排序，所以要放置到下面 可以加写屏障 setExclusiveOwnerThread(null); setState(0); return true; } @Override // 是否持有独占锁 protected boolean isHeldExclusively() { return getState() == 1; } public Condition newCondition() { return new ConditionObject(); } }
自定义锁
// 自定义锁（不可重入锁） class MyLock implements Lock { private 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(); } }
测试一下
public static void main(String[] args) { 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...");
心得
起源
早期程序员会自己通过一种同步器去实现另一种相近的同步器，例如用可重入锁去实现信号量，或反之。这显然不够优雅，于是在 JSR166（java 规范提案）中创建了 AQS，提供了这种通用的同步器机制。
目标
AQS 要实现的功能目标
- 阻塞版本获取锁 acquire 和非阻塞的版本尝试获取锁 tryAcquire
- 获取锁超时机制
- 通过打断取消机制
- 独占机制及共享机制
- 条件不满足时的等待机制
设计
AQS 的基本思想其实很简单
获取锁的逻辑
while(state 状态不允许获取) { if(队列中还没有此线程) { 入队并阻塞 } } 当前线程出队
释放锁的逻辑
if(state 状态允许了) { 恢复阻塞的线程(s) }
要点
- 原子维护 state 状态
- 阻塞及恢复线程
- 维护队列
state 设计
- state 使用 volatile 配合 cas 保证其修改时的原子性
- state 使用了 32bit int 来维护同步状态，因为当时使用 long 在很多平台下测试的结果并不理想
阻塞恢复设计
- 早期的控制线程暂停和恢复的 api 有 suspend 和 resume，但它们是不可用的，因为如果先调用的 resume 那么 suspend 将感知不到
- 解决方法是使用 park & unpark 来实现线程的暂停和恢复，具体原理在之前讲过了，先 unpark 再 park 也没问题
- park & unpark 是针对线程的，而不是针对同步器的，因此控制粒度更为精细
- park 线程还可以通过 interrupt 打断
队列设计
- 使用了 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 好处：
- 无锁，使用自旋
- 快速，无阻塞
CLH队列本身并不会导致线程安全问题。相反，CLH队列是一种用于实现自旋锁等同步机制的数据结构，能够有效地保证线程安全性。
CLH队列（Craig, Landin, and Hagersten queue）是一种基于链表的自旋锁等待队列，它通常应用于自旋锁的实现中。在CLH队列中，每个线程都持有一个自旋锁的状态变量，并通过这些状态变量来构成一个链表结构。当一个线程需要获取锁时，它会将自己的状态设置为“已锁定”，并将自己加入到队列的尾部。然后，它会等待前一个线程释放锁，并检查前一个线程的状态变量，以确定是否可以进入临界区或者继续等待。
由于CLH队列通过显式的状态变量和链表结构来组织线程的等待顺序，因此不会出现像传统锁中的竞争、饥饿等问题。CLH队列的设计使得每个线程按照严格的FIFO顺序等待锁的释放，从而确保了线程安全性。
总之，CLH队列本身并不会导致线程安全问题，它实际上是一种用于保证线程安全的同步机制。然而，在实际使用中，仍然需要注意如何正确地使用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; } } } } 首先，这段代码通过一个无限循环 for (;;) 来进行尾部节点的插入操作。 在每次循环开始时，首先获取当前的尾部节点 t = tail。 如果当前尾部节点为 null，表示队列中还没有元素，这时会尝试初始化队列，将头节点和尾节点都初始化为一个虚拟的哨兵节点（dummy node）。这个虚拟节点并不存储实际的数据，只是作为一个占位符存在，方便后续操作。 如果当前尾部节点不为 null，则将待插入节点的 prev 指针指向当前尾部节点，然后尝试使用CAS操作将尾部节点更新为待插入节点。如果CAS操作成功，表示插入成功，此时需要再次将原尾部节点的 next 指针指向新插入的节点。最后返回原尾部节点，这是因为在CLH队列中，新插入的节点的前驱节点通常需要用到。
ReentrantLock 原理
非公平锁实现原理
细看类图，这里的同步器类是抽象的，它有两个实现，看名字就知道一个是公平的，一个是非公平的
加锁流程
先从构造器开始看，默认为非公平锁实现
public ReentrantLock() { sync = new NonfairSync(); }
从加速 解锁流程开始看
public void lock() { sync.lock(); }
找到非公平锁的实现
static final class NonfairSync extends Sync { private static final long serialVersionUID = 7316153563782823691L; /** * Performs lock. Try immediate barge, backing up to normal * acquire on failure. */ final void lock() { // 下面两行代码其实在模拟自定义锁的时候都用过 // 尝试修改锁，如果修改成功了，就会将owner线程修改 if (compareAndSetState(0, 1)) setExclusiveOwnerThread(Thread.currentThread()); else acquire(1); } protected final boolean tryAcquire(int acquires) { return nonfairTryAcquire(acquires); } }
没有竞争时
第一个竞争出现时
final void lock() { // 下面两行代码其实在模拟自定义锁的时候都用过 // 尝试修改锁，如果修改成功了，就会将owner线程修改 if (compareAndSetState(0, 1)) setExclusiveOwnerThread(Thread.currentThread()); else acquire(1); } 当出现竞争，if失败，进入else public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); } 这里面首先调用了tryAcquire方法，其实就是尝试去加锁，但是下图所示场景，那一定是失败的，此时就会执行acquireQueued方法。
Thread-1 执行了
- CAS 尝试将 state 由 0 改为 1，结果失败
- 进入 tryAcquire 逻辑，这时 state 已经是1，结果仍然失败
- 接下来进入 addWaiter 逻辑，构造 Node 队列
图中黄色三角表示该 Node 的 waitStatus 状态，其中 0 为默认正常状态
Node 的创建是懒惰的
其中第一个 Node 称为 Dummy（哑元）或哨兵，用来占位，并不关联线程
当前线程进入 acquireQueued 逻辑
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; } // 如果获取不到锁，返回false，就会进入到这个if if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } }
acquireQueued 会在一个死循环中不断尝试获得锁，失败后进入 park 阻塞
如果自己是紧邻着 head（排第二位），那么再次 tryAcquire 尝试获取锁，当然这时 state 仍为 1，失败
剑指JUC原理-14.ReentrantLock原理（下）:https://developer.aliyun.com/article/1413650
