阿昌教你看懂AQS核心流程

----

一、预备

Hello!感谢能看菜鸡阿昌的文章 ๑•̀ㅂ•́)و✧

这几天在学习阳哥的AQS，发现他简直就是艺术品。

这里记录一下AQS的核心执行流程。我们以ReentrantLock为例子，进行突破。

因为AQS涉及到Juc的知识，所以这里需要有一些前置知识如下： ≡ω≡

• Juc知识
• LockSupport
• 模版设计模式
• 重入锁/自旋锁
• CAS
• Unsafe

如果不晓得，请自行去补以上的知识，再看一下的内容

也是建议打开IDEA进行源码调试一起食用极佳

---

二、前言

1、什么是AQS

• 字面意思

  • 英文缩写：Abstract Queued Synchronizer
  • 中文直译：抽象队列同步器

• 技术解释

  • AQS = 资源变量State +双向虚拟队列FIFO

• 简单解释

  通过一个资源state，来表示锁的状态。0是自由状态，大于0代表被占用。抢不到资源的线程，就放在FIFO队列中进行管理，争取下次再进行争取获取资源

---

2、那为什么要学AQS呢

Juc线程流程控制&锁的底层实现的基础，如：

• ReentrantLock
• CountDownLatch
• ReentrantReadWriteLock
• Semaphore
• 等…..

---

3、ReentrantLock与AQS的关系

• ReentrantLock类继承实现图

ReentrantLock中存在一个内部类Sync，他继承了AQS，所以ReentrantLock是通过构造内聚Sync类，来间接实现AQS的内容。

• 任何一个Lock接口的实现类，内部都是【聚合】了一个【队列同步器】的子类完成线程访问控制的

---

4、AQS的内部体系关系

下面是我精炼出来的内部体系关系内容：省略了很多

public abstract class AbstractQueuedSynchronizer{//AQS
    private transient volatile Node head;//头节点
    private transient volatile Node tail;//尾节点
    private volatile int state;//资源State变量，默认值为0
    private static final Unsafe unsafe = Unsafe.getUnsafe();//直接操作内存的unsafe工具包
    static final class Node {
        volatile int waitStatus;//线程Node节点状态
        volatile Node prev;//前节点
        volatile Node next;//后节点
        volatile Thread thread;//Node节点的线程
    }
}

上面会看到，AQS中会将线程进行包装成一个Node节点，它有前/后节点。这里就会发现他没有一个真正的双向队列，而是通过包装线程为Node类，并指定里面前后的指向，用前/后节点指针指向来实现一个虚拟的双向队列。

---

5、AQS同步队列的基本结构

---

三、正文

我们这里以两个线程抢资源锁为例，测试代码如下：

public static void main(String[] args) {
    ReentrantLock lock = new ReentrantLock();

    new Thread(()->{
        try {
            lock.lock();
            System.out.println("线程A拿到锁，并执行模拟业务执行60分钟");
            Thread.sleep(60*60*1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    },"线程A").start();

    new Thread(()->{
        try {
            lock.lock();
            System.out.println("线程B拿到锁");
        }finally {
            lock.unlock();
        }
    },"线程B").start();

    System.out.println("程序结束.....");
}

按照main执行流程下来，假设main线程先创建线程A，那么肯定先会执行lock.lock()，让我们看看lock()方法的底层到底做了什么，是如何通过AQS去让线程之间阻塞执行的。

1、线程A开始抢锁

①lock()

进来他先会执行sync的lock()方法

那我们上面已经知道了sync对象是Sync类，内聚变量，一开始并未对齐初始化。

那我们开始创建ReentrantLock对象的时候，就会对Sync进行初始化，以上是以无参构造器举例（默认是非公平锁）

因此，我们上面执行的sync.lock()会去执行Sync的lock()方法，而lock()方法是一个抽象方法

所以就会去执行我们无参构造给初始化的NonfairSync的lock()方法。

首先进入lock()方法，他会进行if-else判断，进行执行compareAndSetState(0,1)

---

②compareAndSetState()

我们看方法的命名，就知道他必然是CAS方法，他肯定是调用unsafe工具包直接操作内存地址的。

//CAS，预期的的值是0，想要更新为1
protected final boolean compareAndSetState(int expect, int update) {//0,1
    //this：AQS队列同步器
    //stateOffset：资源变量，stateOffset内存的偏移量，也就是地址
    //expect：期盼值是0，他是默认值是0，所以肯定是为0
    //update：更新值是1
    return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

如果此时队列同步器AQS的state资源变量的值是0，那么就更新为1，且该操作是原子操作。再上面，我们说了state资源变量的默认值为0，那这里的CAS操作必然是成功的。就会修改完成后返回true。

compareAndSetState(0, 1)返回的是true，所以就会继续往下执行setExclusiveOwnerThread(Thread.currentThread())

---

③setExclusiveOwnerThread()

看这个名字都能看出来，他是设置排他Exclusive的拥有者线程是哪个

设置占有线程，逻辑也很简单，就是将Thread exclusiveOwnerThread变量设置为当前线程，代表此时的队列同步器处理器占有的线程的是此时进来的线程A。

那执行完，这里的lock()逻辑就完毕返回了，接下来就会执行业务逻辑，这里我们模拟是业务执行过长时间，线程A一直持有锁。

那么此时的全局状态图应该是如下：

线程A去强占资源，state资源变量通过CAS操作从0修改为1。setExclusiveOwnerThread()设置了当前AQS同步队列的占有线程为线程A

---

2、线程B开始抢锁

线程B肯定有会像上面的线程A一样去执行lock()方法

①lock()

同样，他会执行到lock()方法，也会去先执行compareAndSetState()。

但是此时传入的参数为0,1。即期望为0，更新为1的操作，但是此时state变量已经被占用了，且因为线程A被CAS修改为1，那么结果很不意外的，compareAndSetState必然返回false，则就不会进入setExclusiveOwnerThread去改变此时AQS的拥有线程，因为此时线程A正在占用。

那么，他就会去执行acquire(1)的方法

---

②acquire()

因为上面，直接CAS操作失败了，那么就会执行acquire。

上面会看到3个重要方法：

1. tryAcquire()
2. addWaiter()
3. acquireQueued()

---

③tryAcquire()

一点开，会发现，他上来就给抛了个UnsupportedOperationException异常，那就会很明显的看出来，这里使用的模版设计模式，必须要实现抽象的AbstractQueuedSynchronizer的tryAcquire方法。因为我们用的默认的空参构造器，所以初始化Sync用的是NonfairSync。进入之后他就会执行nonfairTryAcquire()

---

④nonfairTryAcquire()

final boolean nonfairTryAcquire(int acquires) {
    //拿到当前线程引用，这里会拿到线程B
    final Thread current = Thread.currentThread();
    //拿到此时state资源的值，此时state的值为1，被线程A给占用CAS修改为1
    int c = getState();
    // c = 1，不进入该逻辑
    if (c == 0) {
        if (compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    //判断是否是可重入锁
    //判断current当前是否是占用的线程，也就是线程A，那必然不是，此时是线程B来抢资源
    //如果线程A因为某些原因放弃锁或再去获取同一把锁，这里就会进入
    else if (current == getExclusiveOwnerThread()) {
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    //因为 c = 1，上面都没有执行，所以最终返回false
    return false;
}

因为此时state的状态为1，那么线程B在tryAcquire()什么都不执行，并返回false，因为外面有一个!非，所以&&会继续向有执行，也就是会执行acquireQueued(addWaiter(Node.EXCLUSIVE), arg)

---

⑤addWaiter()

传入Node模式为EXCLUSIVE，排他类型

private Node addWaiter(Node mode) {
    //将线程B包装成一个Node节点，模式为EXCLUSIVE排他
    Node node = new Node(Thread.currentThread(), mode);
    //pred节点为tail尾节点，但是此时尾节点的指针肯定是null
    Node pred = tail;
    //所以Node pred = null
    //进不来进行的if
    if (pred != null) {
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    //那么就会走到这里，进行enq将线程B入队
    enq(node);
    return node;
}

---

⑥enq()

进行enq将线程B入队

上来看到是一个自旋for (;;)

private Node enq(final Node node) {
    //自旋，死循环
    for (;;) {
        //此时的tail尾节点为null
        //Node t = null
        Node t = tail;
        //进入如下逻辑
        if (t == null) {
			//初始化傀儡节点new Node()，一个空的Node，作为占位
            if (compareAndSetHead(new Node()))
                //尾节点设置为头节点指向，也就是傀儡节点
                tail = head;
        } else {
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

• compareAndSetHead

CAS操作，将当前队列同步器的头节点内存偏移量headOffset，预期是null，设置为上面new Node傀儡节点空Node。

• 此时全局状态图

经过这里给AQS同步队列初始化了一个空Node/傀儡节点，因为这里是自旋的for (;;)，所以线程B还会再循环再执行。

private Node enq(final Node node) {
    //自旋第二次进来
    for (;;) {
        //此时tail尾节点的指向是傀儡节点，所以t不等于null
        Node t = tail;
        if (t == null) { 
            if (compareAndSetHead(new Node()))
                tail = head;
            //进入下面的else逻辑，此时这里才为线程B的Node进行入队
        } else {
            //node为线程B的Node对象
            //设置线程B的Node对象的prev前节点为t，t就是傀儡节点
            node.prev = t;
            //CAS设置tail尾节点为线程B节点
            if (compareAndSetTail(t, node)) {
                //t傀儡节点的next后节点，指向node线程B
                t.next = node;
                return t;
            }
        }
    }
}

• 以上队列状态图如下

执行完enq()，就会返回线程B的node节点地址

那么接下来就会执行acquireQueued()，他传入的参数1为返回的线程B的node节点，和上面传下来的arg=1的参数。

---

⑦acquireQueued()

acquireQueued最为关键，就是将线程B抢不到锁后阻塞挂起

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        //自旋，死循环
        for (;;) {
            //拿到线程B的node节点的前节点给p对象
            final Node p = node.predecessor();
            // 判断线程B的前节点是否是头节点，根据上面的全局状态图，我们可以知道现在的头节点的指向是空节点，也就是傀儡节点
            //p是傀儡节点，他是头节点，但不会进入，因为tryAcquire会再次抢锁，内容跟上面的一样，因为线程A一直占用着资源，所以不进入逻辑
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            //判断此时线程B的Node的waitStatus状态
            //p：线程B节点的前节点指向，也就是傀儡节点
            //node：线程B节点
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

• shouldParkAfterFailedAcquire

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    //拿到傀儡节点的waitStates状态，因为是类的成员变量，所以初始化默认值为0
    int ws = pred.waitStatus;
    //判断傀儡节点的waitStates是否等于 -1
    if (ws == Node.SIGNAL)
        return true;
    //傀儡节点的waitStates = 0
    if (ws > 0) {
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        //执行这里，通过CAS操作，将傀儡节点的waitStates设置为-1
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    //返回false
    return false;
}

因为这里是自旋的死循环，所以还会再次进入执行。

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        //第二次进入
        for (;;) {
            //拿到线程B的前节点指向，也就是傀儡节点
            //p为傀儡节点指向
            final Node p = node.predecessor();
            //同样不进入如下逻辑，模拟线程A还在占用资源并未结束
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            //再次进入shouldParkAfterFailedAcquire
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

• 全局状态图

因为我们在shouldParkAfterFailedAcquire给p节点，也就是傀儡节点设置了他的waitState为Node.SIGNAL，也就是-1

第二次进入再次判断傀儡节点的waitStatus，因为上面设置为了-1，所以这次就会返回了true。那么接下来就会执行parkAndCheckInterrupt

• parkAndCheckInterrupt()

出现了！！！这里就是让线程B阻塞的罪魁祸首，LockSupport的park，这个this就是线程B。那么这里线程B就需要有人执行LockSupport.unpark(线程B)来唤醒他，不然他就会永远被阻塞挂起。

• 全局状态图

接下来就是等待持有资源的线程A执行unlock()方法了

---

3、线程A放弃锁

线程A执行玩逻辑之后，最终finally就会执行lock.unlock()，进行解锁，放弃锁资源

这里也能看出，ReentrantLock的unlock方法，依然是调用sync的release方法

会看到这里他会先进入if，去执行tryRelease，同样，这个tryRelease执行的必然是实现类的方法，也就是Sync的tryRelease，模版设计模式

---

①tryRelease

protected final boolean tryRelease(int releases) {
    //getState()获取当前资源state值，看上面的全局状态，此时为1
    //releases为传入的值1
    //c = 1 -1 
    //c = 0
    int c = getState() - releases;
    //判断当前线程是否是资源占用线程
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    //进入如下逻辑
    if (c == 0) {
        free = true;
        //将当前占用资源的线程改为null
        setExclusiveOwnerThread(null);
    }
    //并设置当前state资源变量为c，也就是0
    setState(c);
    return free;
}

• 全局状态图

public final boolean release(int arg) {
    //上面返回的free为true，进入下面逻辑
    if (tryRelease(arg)) {
        //拿到头节点的引用并赋给h变量，头节点是傀儡节点
        //h为傀儡节点
        Node h = head;
        //头节点为傀儡节点不为null，且傀儡节点的waitState状态为-1，不等于0
        if (h != null && h.waitStatus != 0)
            //执行unparkSuccessor，唤醒阻塞的线程
            unparkSuccessor(h);
        return true;
    }
    return false;
}

---

②unparkSuccessor

这个方法会唤醒，上面阻塞的线程B

private void unparkSuccessor(Node node) {
    //拿到傀儡节点的waitStatus状态
    //ws = -1
    int ws = node.waitStatus;
    //进入逻辑
    if (ws < 0)
        //CAS，设置傀儡节点的waitStatus为0
        compareAndSetWaitStatus(node, ws, 0);
    //拿到傀儡节点的后节点，也就是线程B节点
    //s为线程B节点
    Node s = node.next;
    //此时s为线程B节点不为null，且线程B节点的waitStatus为0，不进入逻辑
    if (s == null || s.waitStatus > 0) {
        s = null;
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
    //s为线程B节点不为null，进入逻辑
    if (s != null)
        //为线程B唤醒
        LockSupport.unpark(s.thread);
}

LockSupport.unpark(s.thread)为线程B唤醒，所以线程B就是开始执行。这里线程A的unlock()就执行完毕了

• 全局状态图

---

4、线程B被线程A唤醒，并尝试去再抢锁

线程B被线程A唤醒，并自旋死循环，再次强占锁资源

①acquireQueued

final boolean acquireQueued(final Node node, long arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        //被线程A唤醒的线程B再次进入
        for (;;) {
            //拿到线程B的前节点，也就是傀儡节点
            //p为傀儡节点
            final Node p = node.predecessor();
            //此时傀儡节点为头节点，且再次执行tryAcquire去强占资源，此时会返回true，并强占成功
            if (p == head && tryAcquire(arg)) {
                //设置头节点为线程B
                //设置线程B的Node中的线程为null
                //线程B的Node的前节点为null
                setHead(node);
                //将傀儡节点的后节点设置为null
                p.next = null; 
                failed = false;
                //返回是否被中断，默认是没被中断也是false
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

• tryAcquire()

同样是模版设计模式，会去执行Sync的实现

• nonfairTryAcquire()

final boolean nonfairTryAcquire(int acquires) {
    //拿到当前线程B
    final Thread current = Thread.currentThread();
    //获取当前的资源State变量，此时为0，因为已经被线程A的unlock方法设置为了0
    int c = getState();
    //进入逻辑
    if (c == 0) {
        //将资源State变量设置为传入的acquires，也就是1
        if (compareAndSetState(0, acquires)) {
            //再设置当前资源占用的线程为当前线程，也就是线程B
            setExclusiveOwnerThread(current);
            //返回true
            return true;
        }
    }
    else if (current == getExclusiveOwnerThread()) {
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

• setHead()

设置线程B的Node为头节点，设置线程B的Node中的线程为null，线程B的Node的前节点为null

• 全局状态图

以此循环，如果有别的线程加入队列也是维持以上的流转管理操作。

四、结尾

以上就满足了AQS的管理流转，通过资源变量State+AQS的同步管理队列，实现了线程强占资源的管理。

用资源变量State来控制当前资源是否被占用，0为未占用状态，大于0为被占用情况。

抢不到资源的线程会被AQS维护管理在虚拟的同步管理队列，通过LockSupprot进行线程的唤醒和阻塞。

现在再细品一开始的话：

通过一个资源state，来表示锁的状态。0是自由状态，大于0代表被占用。抢不到资源的线程，就放在FIFO队列中进行管理，争取下次再进行争取获取资源

以上就是这次的全部内容，感谢你能看到这里(๑ˉ∀ˉ๑)!!!