AQS - 从干饭角度解析

## AQS 是什么？

AQS （AbstractQueuedSynchronizer）抽象队列同步器，是JUC并发包下多种锁的底层实现。如CountDownLatch、Semaphore、ReentrantReadWriteLock。

其底层是CLH（Craig, Landin, and Hagersten）锁的改进版。 核心原理是一样的，把竞争的线程排成先入先出队列，排队的节点只监听pre节点的锁状态。

```java
public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer {
    // 队列头节点
    private transient volatile Node head;

// 队列尾节点
    private transient volatile Node tail;

// 同步资源状态，已排队的人盯着前面就行，这里作用就是饭还剩XX份。给新来的人看要不要加入排队的。
    private volatile int state;

// cas 原子操作使用unsafe，直接调用系统提供的能力
    private static final Unsafe unsafe = Unsafe.getUnsafe();

/* 
    * CLH的变种
    *        +------+  prev +-----+       +-----+
    *   head |      | <---- |     | <---- |     |  tail
    *        +------+       +-----+       +-----+
    */
    static final class Node {
        // nextWaiter = SHARED 时，说明是共享锁
        static final Node SHARED = new Node();
        // nextWaiter = EXCLUSIVE 时，说明是独占锁
        static final Node EXCLUSIVE = null;

// 线程状态。干饭人的状态
        volatile int waitStatus;
        // 前驱节点，用于监听他的waitStatus变化以判断是否能获取到锁
        volatile Node prev;
        // 后置节点，用于解锁时，通知后置的节点。吃饭了通知后面发呆的同学
        volatile Node next;
        // 申请锁的线程信息，干饭人
        volatile Thread thread;
        // 共享锁才有用，指向下个等待的node节点=SHARED
        Node nextWaiter;
    }
    
    //独占方式。尝试获取资源，成功则返回true，失败则返回false。
    protected boolean tryAcquire(int)
    //独占方式。尝试释放资源，成功则返回true，失败则返回false。
    protected boolean tryRelease(int)
    //共享方式。尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。
    protected int tryAcquireShared(int)
    //共享方式。尝试释放资源，成功则返回true，失败则返回false。
    protected boolean tryReleaseShared(int)
    //该线程是否正在独占资源。只有用到condition才需要去实现它。
    protected boolean isHeldExclusively()
}
```

## 为什么我们要用AQS ?

AQS 没诞生之前，我们也可以使用lock和unlock一个对象，解决并发时原子性、有序性、可见性。 但是当竞争线程达到成千上万时，一窝蜂地争抢唯一的对象就会大大影响性能（线程切换性能下降），还有可能一直抢不到一致不处理形成假死现象（锁饥饿问题）。从干饭人角度看，就是如果是一窝蜂抢饭吃，大家都向系统申请干饭，系统处理不过来。部分同学长期抢不到还会饿死。

## AQS 怎么做的？

AQS 希望这些成千上万的线程排好队，一个一个来争抢资源。 当一个线程好了通知下一个线程，形成一个先进先出队列 FIFO( first-in-first-out)。

这个结构最早就是CLH， 传统CLH 就像食堂排队打饭，每个客户（线程节点）来了，从队尾tail开始排队。排队的客户都只盯着前面那个客户好了没，不需要都来问食堂大妈（系统）好了没。

而传统CLH也有自己的问题，在排队等待时，CLH用了自旋监听前一个客户的状态时，也就是死循环查询，查到为止。而 AQS 通过记录后置节点next，变成了信号通知（LockSupport的park和unpark函数），优化了性能。当前客户（线程节点）处理完要负责转过身叫醒后面发呆等待的排队客户，十分有爱。

同时还有一些争对GC优化，出队列时设置为null； 细化waitStatus等待状态，提供独占、共享、条件队列等功能；显示维护双向队列，提供超时、取消功能等等

当然魔鬼在细节，关于他的底层原理，推荐阅读https://tech.meituan.com/2019/12/05/aqs-theory-and-apply.html、 https://mp.weixin.qq.com/s/jEx-4XhNGOFdCo4Nou5tqg.

## AQS 怎么使用？

AQS定义了两种使用模式， SHARED共享模式（多个线程执行）、EXCLUSIVE独占模式（只有一个线程能执行）。
共享模式实现 tryAcquireShared和tryReleaseShared；独占模式实现 tryAcquire、tryReleaseShared。特殊的写独占读共享的ReentrantReadWriteLock，就是都实现掉。

以 CountDownLatch 为例，他是共享锁。一般是主线程调用CountDownLatch::wait进行等待,  子线程CountDownLatch::countDown，消耗一份资源。 当定义的资源总数为0表示所有的线程都已经完成，然后再结束主线程的等待。应用场景还是以干饭为例： 大家都吃完饭了再干活

```
public class CountDownLatch {
	// 关键数据结构继承了AQS
    private static final class Sync extends AbstractQueuedSynchronizer {
       
        Sync(int count) {
        	// 定义资源count，这里限制了只有创建时才能初始化一次资源。大家要吃几份饭
            setState(count);
        }

// 实现扩展接口，是否还有资源干饭。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源
        protected int tryAcquireShared(int acquires) {
            return (getState() == 0) ? 1 : -1;
        }

// 实现扩展接口，尝试消耗资源吃饭，成功则返回true，失败则返回false。
        protected boolean tryReleaseShared(int releases) {
            // cas 进行count-1操作，count=0，消耗失败。
			// 由于count代表整个队列的state状态，如果在尝试时就减一维护状态
            // ... 省略具体代码
        }
    }

private final Sync sync;
	
	public void await() throws InterruptedException {
        // 主线程初始化做好饭后，等着大家干完饭。等待await结束就干活。
        sync.acquireSharedInterruptibly(1);
    }
    
}

```

我们补上AQS的函数执行细节
```java
public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer {
	
	public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            // 有人掀桌子，不干了，抛异常
            throw new InterruptedException();
        // 如果还有资源可用，大家饭没吃完
        if (tryAcquireShared(arg) < 0)
            // 那就等大家把饭吃完
            doAcquireSharedInterruptibly(arg);
    }
	
	private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
        // 创建当前线程新建为共享节点，并排到队尾等待
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            // 用于持续尝试获取锁直到成功或者被中断，一直盯着
            for (;;) {
                // 获取前置节点，排队时盯着前面的人
                final Node p = node.predecessor();
                // 前置为head，排到了，开始处理
                if (p == head) {
                    // CountDownLatch定义tryAcquireShared：state=0才运行放行，必须大家吃完
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
                        // 还有资源则设置新的头节点，并根据获取的资源数量决定是否传播锁
                        // setHeadAndPropagate 后续调用了LockSupport.unpark操作唤醒下个节点
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                } else {
                    // 继续循环盯着前面的人，直到排到为止
                }
                // 检查是否中断，设置线程等待并检查中断
                // parkAndCheckInterrupt 调用了LockSupport::park进行等待操作，不是完全自旋
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
            // 出现问题，那就取消尝试获取锁并离开排队队列
            if (failed)
                cancelAcquire(node);
        }
    }
}
```