AQS原理 学习笔记

Posted 2021-07-20 Shi Peng

一、什么是AQS

AQS即AbstractQueuedSynchronizer，是一个构建锁和同步器的框架。

AQS支持独占锁（exclusive）和共享锁（share）两种模式：

• 独占锁：只能被一个线程获取到，如ReentrantLock
• 共享锁：可以被多个线程同时获取，如CountDownLatch

无论是独占锁还是共享锁，本质上都是对AQS内部的一个变量state的获取。state是一个原子的int变量，用来表示锁的状态、资源数等。

二、AQS内部的数据结构及原理

同步队列的作用：当线程获取资源失败后，就进入同步队列的尾部，保持自旋等待，不断判断自己是否是链表的头结点，如果是，就不断尝试获取资源，获取成功后，则退出同步队列。

条件队列是为Lock实现的一个基础同步器，并且一个线程可能会有多个条件队列，只有在使用了Condition才会存在条件队列。

同步队列和条件队列，都是用一个个Node组成。AQS内部有一个静态内部类Node：

    static final class Node {
        static final Node EXCLUSIVE = null;

        //当前节点由于超时或中断被取消
        static final int CANCELLED =  1;
     
        //表示当前节点的前节点被阻塞
        static final int SIGNAL    = -1;
        
        //当前节点在等待condition
        static final int CONDITION = -2;
      
        //状态需要向后传播
        static final int PROPAGATE = -3;
        
        volatile int waitStatus;
        
        volatile Node prev;
        volatile Node next;
        volatile Thread thread;

        Node nextWaiter;

        final boolean isShared() {
            return nextWaiter == SHARED;
        }

        final Node predecessor() throws NullPointerException {
            Node p = prev;
            if (p == null)
                throw new NullPointerException();
            else
                return p;
        }

        Node() {    // Used to establish initial head or SHARED marker
        }

        Node(Thread thread, Node mode) {     // Used by addWaiter
            this.nextWaiter = mode;
            this.thread = thread;
        }

        Node(Thread thread, int waitStatus) { // Used by Condition
            this.waitStatus = waitStatus;
            this.thread = thread;
        }
    }

最重要的方法：

//独占模式下获取资源
public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

private Node addWaiter(Node mode) {
        Node node = new Node(Thread.currentThread(), mode);
        // Try the fast path of enq; backup to full enq on failure
        Node pred = tail;
        //尝试快速入队
        if (pred != null) { //队列已经初始化
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node; //快速入队成功后，就直接返回了
            }
        }
        //快速入队失败，也就是说队列都还每初始化，就使用enq
        enq(node);
        return node;
    }
    
    //执行入队
     private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            if (t == null) { // Must initialize
            //如果队列为空，用一个空节点充当队列头
                if (compareAndSetHead(new Node()))
                    tail = head;//尾部指针也指向队列头
            } else {
                //队列已经初始化，入队
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;//打断循环
                }
            }
        }
    }

 final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();//拿到node的上一个节点
                //前置节点为head，说明可以尝试获取资源。排队成功后，尝试拿锁
                if (p == head && tryAcquire(arg)) {
                    setHead(node);//获取成功，更新head节点
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                //尝试拿锁失败后，根据条件进行park
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }
    //获取资源失败后，检测并更新等待状态
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            /*
             * This node has already set status asking a release
             * to signal it, so it can safely park.
             */
            return true;
        if (ws > 0) {
            /*
             * Predecessor was cancelled. Skip over predecessors and
             * indicate retry.
             */
            do {
            //如果前节点取消了，那就往前找到一个等待状态的接待你，并排在它的后面
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            /*
             * waitStatus must be 0 or PROPAGATE.  Indicate that we
             * need a signal, but don't park yet.  Caller will need to
             * retry to make sure it cannot acquire before parking.
             */
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }
    //阻塞当前线程，返回中断状态
    private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this);
        return Thread.interrupted();
    }

三、从ReentrantLock看AQS的原理

3.1、前言

AQS是一种提供了原子式管理同步状态、阻塞和唤醒线程功能，以及队列模型的简单应用框架。下面姜通过ReentrantLock与AQS的关联，来深入解读AQS相关独占锁的知识点。

ReentrantLock和Synchronized对比：

通过伪代码对比：

// **************************Synchronized的使用方式**************************
// 1.用于代码块
synchronized (this) {}
// 2.用于对象
synchronized (object) {}
// 3.用于方法
public synchronized void test () {}
// 4.可重入
for (int i = 0; i < 100; i++) {
	synchronized (this) {}
}
// **************************ReentrantLock的使用方式**************************
public void test () throw Exception {
	// 1.初始化选择公平锁、非公平锁
	ReentrantLock lock = new ReentrantLock(true);
	// 2.可用于代码块
	lock.lock();
	try {
		try {
			// 3.支持多种加锁方式，比较灵活; 具有可重入特性
			if(lock.tryLock(100, TimeUnit.MILLISECONDS)){ }
		} finally {
			// 4.手动释放锁
			lock.unlock()
		}
	} finally {
		lock.unlock();
	}
}

ReentrantLock和AQS的关联

ReentrantLock支持公平锁和非公平锁，并且ReentrantLock的底层就是由AQS来实现的。那么ReentrantLock是如何通过公平锁和非公平锁与AQS关联起来呢？

非公平锁加锁源码如下：

// java.util.concurrent.locks.ReentrantLock#NonfairSync

// 非公平锁
static final class NonfairSync extends Sync {
	...
	final void lock() {
		if (compareAndSetState(0, 1))
			setExclusiveOwnerThread(Thread.currentThread());
		else
			acquire(1);
		}
     ...
}

这块代码的含义为：

若通过CAS设置变量State（同步状态）成功，也就是获取锁成功，则将当前线程设置为独占线程。

若通过CAS设置变量State（同步状态）失败，也就是获取锁失败，则进入Acquire方法进行后续处理。

第一步很好理解，但第二步获取锁失败后，后续的处理策略是怎么样的呢？这块可能会有以下思考：

某个线程获取锁失败的后续流程是什么呢？有以下两种可能：
(1) 将当前线程获锁结果设置为失败，获取锁流程结束。这种设计会极大降低系统的并发度，并不满足我们实际的需求。所以就需要下面这种流程，也就是AQS框架的处理流程。
(2) 存在某种排队等候机制，线程继续等待，仍然保留获取锁的可能，获取锁流程仍在继续。

对于问题1的第二种情况，既然说到了排队等候机制，那么就一定会有某种队列形成，这样的队列是什么数据结构呢？

处于排队等候机制中的线程，什么时候可以有机会获取锁呢？

如果处于排队等候机制中的线程一直无法获取锁，还是需要一直等待吗，还是有别的策略来解决这一问题？

带着非公平锁的这些问题，再看下公平锁源码中获锁的方式：

// java.util.concurrent.locks.ReentrantLock#FairSync

static final class FairSync extends Sync {
  ...  
	final void lock() {
		acquire(1);
	}
  ...
}

看到这块代码，我们可能会存在这种疑问：Lock函数通过Acquire方法进行加锁，但是具体是如何加锁的呢？

结合公平锁和非公平锁的加锁流程，虽然流程上有一定的不同，但是都调用了Acquire方法，而Acquire方法是FairSync和UnfairSync的父类AQS中的核心方法。

对于上边提到的问题，其实在ReentrantLock类源码中都无法解答，而这些问题的答案，都是位于Acquire方法所在的类AbstractQueuedSynchronizer中，也就是本文的核心——AQS。

3.2、AQS

AQS框架：

上图中有颜色的为Method，无颜色的为Attribution。

总的来说，AQS框架共分为五层，自上而下由浅入深，从AQS对外暴露的API到底层基础数据。

当有自定义同步器接入时，只需重写第一层所需要的部分方法即可，不需要关注底层具体的实现流程。当自定义同步器进行加锁或者解锁操作时，先经过第一层的API进入AQS内部方法，然后经过第二层进行锁的获取，接着对于获取锁失败的流程，进入第三层和第四层的等待队列处理，而这些处理方式均依赖于第五层的基础数据提供层。

下面我们会从整体到细节，从流程到方法逐一剖析AQS框架，主要分析过程如下：

3.2.1、AQS原理概述

AQS核心思想是，如果被请求的共享资源空闲，那么就将当前请求资源的线程设置为有效的工作线程，将共享资源设置为锁定状态；如果共享资源被占用，就需要一定的阻塞等待唤醒机制来保证锁分配。这个机制主要用的是CLH队列的变体实现的，将暂时获取不到锁的线程加入到队列中。

CLH：Craig、Landin and Hagersten队列，是单向链表，AQS中的队列是CLH变体的虚拟双向队列（FIFO），AQS是通过将每条请求共享资源的线程封装成一个节点来实现锁的分配。

主要原理如下：

AQS使用一个Volatile的int类型的成员变量来表示同步状态，通过内置的FIFO队列来完成资源获取的排队工作，通过CAS完成对State值的修改。

3.2.1.1、AQS的数据结构

先来看下AQS中最基本的数据结构——Node，Node即为上面CLH变体队列中的节点。

解释一下几个方法和属性值的含义：

线程两种锁的模式：

waitStatus有下面几个枚举值：

3.2.1.2、同步状态State

AQS中维护了一个名为state的字段，意为同步状态，是由Volatile修饰的，用于展示当前临界资源的获锁情况。

// java.util.concurrent.locks.AbstractQueuedSynchronizer

private volatile int state;

下面提供了几个访问这个字段的方法：

这几个方法都是Final修饰的，说明子类中无法重写它们。我们可以通过修改State字段表示的同步状态来实现多线程的独占模式和共享模式（加锁过程）。

对于我们自定义的同步工具，需要自定义获取同步状态和释放状态的方式，也就是AQS架构图中的第一层：API层。

更多详情：https://tech.meituan.com/2019/12/05/aqs-theory-and-apply.html