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

剑指JUC原理-14.ReentrantLock原理（上）