运维首页 > 开发运维 > 三分钟带你搞懂 AQS 原理设计

三分钟带你搞懂 AQS 原理设计

2024年 1月 12日 41.1k 0

一、摘要

在之前的文章中，我们介绍了 ReentrantLock、ReadWriteLock、CountDownLatch、CyclicBarrier、Semaphore、ThreadPoolExecutor 等并发工具类的使用方式，它们在请求共享资源的时候，都能实现线程同步的效果。

在使用方式上稍有不同，有的是独占式，多个线程竞争时只有一个线程能执行方法，比如 ReentrantLock 等；有的是共享式，多个线程可以同时执行方法，比如：ReadWriteLock、CountDownLatch、Semaphore 等，不同的实现争用共享资源的方式也不同。

如果仔细阅读源码，会发现它们都是基于AbstractQueuedSynchronizer这个抽象类实现的，我们简称 AQS。

AQS 是一个提供了原子式管理同步状态、阻塞和唤醒线程功能的框架，是除了 Java 自带的synchronized关键字之外的锁实现机制。

可以这么说，AQS是JUC包下线程同步类的基石，也是很多面试官喜欢提问的话题，掌握AQS原理对我们深入理解线程同步技术有着非常重要的意义。

本文以ReentrantLock作为切入点，来解读AQS相关的知识点，最后配上简单的应用示例来帮助大家理解 AQS，如果有描述不对的地方，欢迎大家留言指出，不胜感激！

二、ReentrantLock

在之前的线程系列文章中，我们介绍了ReentrantLock的基本用法，它是一个可重入的互斥锁，它具有与使用synchronized关键字一样的效果，并且功能更加强大，编程更加灵活，支持公平锁和非公平锁两种模式。

使用方式也非常简单，只需要在相应的代码上调用加锁和释放锁方法即可，简单示例如下！

public class Counter {

    // 默认非公平锁模式
    private final Lock lock = new ReentrantLock();

    public void add() {
        // 加锁
        lock.lock();
        try {
            // 具体业务逻辑...
        } finally {
            // 释放锁
            lock.unlock();
        }
    }
}

如果阅读lock()和unlock()方法，会发现它的底层都是由AQS来实现的。

下面，我们一起来看看这两个方法的源码实现，本文源码内容摘取自 JDK 1.8 版本，可能不同的版本略有区别！

2.1、lock 方法源码

public class ReentrantLock implements Lock, java.io.Serializable {
    
    // 同步锁实现类
    private final Sync sync;

    public ReentrantLock() {
        // 默认构造方法为非公平锁实现类
        sync = new NonfairSync();
    }

    public ReentrantLock(boolean fair) {
        // true：公平锁实现类，false：非公平锁实现类
        sync = fair ? new FairSync() : new NonfairSync();
    }

    public void lock() {
        // 加锁操作
        sync.lock();
    }

    // 非公平锁实现类
    static final class NonfairSync extends Sync {

         // 加锁操作
        final void lock() {
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
                acquire(1);
        }
    }

    // 公平锁实现类
    static final class FairSync extends Sync {

        // 加锁操作
        final void lock() {
            acquire(1);
        }
    }

    // 公平锁和非公平锁，都继承自 AQS
    abstract static class Sync extends AbstractQueuedSynchronizer {

        // lock 抽象方法
        abstract void lock();
    }
}

从源码上可以清晰的看到，当初始化ReentrantLock对象时，需要指定锁的模式。

默认构造方法是非公平锁模式，采用的是NonfairSync内部实现类；公平锁模式下，则采用的是FairSync内部实现类；这两个内部实现类都继承了Sync抽象类；同时，Sync也继承了AbstractQueuedSynchronizer，也就是我们上文提到的AQS。

如果把lock()方法的请求链路进行抽象，可以用如下图进行简要概括。

图片图片

无论是非公平锁模式还是公平锁模式，可能最终都会调用AQS的acquire()方法，它表示通过独占式的方式加锁，我们继续往下看这个方法的源码，部分核心代码如下：

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {

    // 通过独占式的方式加锁
    public final void acquire(int arg) {
        // 尝试加锁，会回调具体的实现类
        if (!tryAcquire(arg) &&
            // 如果尝试加锁失败，将当前线程加入等待队列
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

    // 由子类完成加锁逻辑的实现，支持重写该方法
    protected boolean tryAcquire(int arg) {
        throw new UnsupportedOperationException();
    }
}

从AQS的源码上可以看出，acquire()方法并不进行具体加锁逻辑的实现，而是通过具体的实现类重写tryAcquire()方法来完成加锁操作，如果加锁失败，会将当前线程加入等待队列。

如果是非公平锁模式，会回调ReentrantLock类的NonfairSync.tryAcquire()方法；如果是公平锁模式，会回调ReentrantLock类的FairSync.tryAcquire()方法，我们继续回看ReentrantLock类的源码。

非公平锁NonfairSync静态内部实现类，相关的源码如下！

// 非公平锁实现类
static final class NonfairSync extends Sync {

     // 加锁操作
    final void lock() {
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }

    // 尝试非公平方式加锁，重写父类 tryAcquire 方法
    protected final boolean tryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {
            // 采用CAS方式修改线程同步状态，如果成功返回true
            if (compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        else if (current == getExclusiveOwnerThread()) {
            // 支持当前线程，重复获得锁，将state值加1
            int nextc = c + acquires;
            if (nextc < 0)
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false;
    }

}

公平锁FairSync静态内部实现类，相关的源码如下！

// 公平锁实现类
static final class FairSync extends Sync {

    // 加锁操作
    final void lock() {
        acquire(1);
    }

    // 尝试公平方式加锁，重写父类 tryAcquire 方法
    protected final boolean tryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {
            // 1）判断等待队列是否有线程处于等待状态，如果没有，尝试获取锁；如果有，就进入等待队列
            // 2）采用CAS方式修改线程同步状态，如果成功返回true
            if (!hasQueuedPredecessors() &&
                compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        else if (current == getExclusiveOwnerThread()) {
            // 支持当前线程，重复获得锁，将state值加1
            int nextc = c + acquires;
            if (nextc < 0)
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false;
    }
}

从源码上可以清晰的看到，无论是是公平锁还是非公平锁模式，都是采用compareAndSetState()方法（简称CAS）进行加锁，如果成功就返回true；同时支持当前线程重复获得锁，也就是之前提到的锁可重入机制。

唯一的区别在于：公平锁实现类多了一个hasQueuedPredecessors()方法判断，它的用途是判断等待队列是否有线程处于等待状态，如果没有，尝试获取锁；如果有，就将当前线程存入等待队列，依此排队，从而保证线程通过公平方式获取锁的目的。

关于 CAS 实现原理，在之前的并发原子类文章中已经有所介绍，通过它加上volatile修饰符可以实现一个无锁的线程安全访问操作，本文不再重复解读，有兴趣的朋友可以翻阅之前的文章。

2.2、unlock 方法源码

public class ReentrantLock implements Lock, java.io.Serializable {

    // 同步锁实现类
    private final Sync sync;

    public void unlock() {
        // 释放锁操作
        sync.release(1);
    }
}

unlock()方法的释放锁实现相对来说就简单多了，整个请求链路可以用如下图进行简要概括。

当调用unlock()方法时，会直接跳转到AQS的release()方法上，AQS相关的源码如下！

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {

    // 释放锁操作
    public final boolean release(int arg) {
        // 尝试释放锁
        if (tryRelease(arg)) {
            // 从队列头部中获取一个等待线程，并进行唤醒操作
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

    // 由子类完成释放锁逻辑的实现，支持重写该方法
    protected boolean tryRelease(int arg) {
        throw new UnsupportedOperationException();
    }
}

与加锁操作类似，AQS的release()方法并不进行具体释放锁逻辑的实现，而是通过具体的实现类重写tryRelease()方法来完成释放锁操作，如果释放锁成功，会从队列头部中获取一个等待线程，并进行唤醒操作。

我们继续回看ReentrantLock类的Sync.tryRelease()释放锁方法，部分核心源码如下：

abstract static class Sync extends AbstractQueuedSynchronizer {

    // 尝试释放锁
    protected final boolean tryRelease(int releases) {
        // 将state值进行减1操作
        int c = getState() - releases;
        if (Thread.currentThread() != getExclusiveOwnerThread())
            throw new IllegalMonitorStateException();
        boolean free = false;
        if (c == 0) {
            free = true;
            setExclusiveOwnerThread(null);
        }
        setState(c);
        return free;
    }
}

相比加锁过程，释放锁要简单的多，主要是将线程的同步状态值进行自减操作。

三、AQS 原理浅析

如果仔细的研究 AQS 的源码，尽管实现上很复杂，但是也有规律可循。

从上到下，整个框架可以分为五层，架构可以用如下图来描述！（图片来自ReentrantLock 的实现看 AQS 的原理及应用 - 美团技术团队）

图片图片

当有自定义线程同步器接入AQS时，只需要按需重写第一层的方法即可，不需要关心底层的实现。

以加锁为例，当调用AQS的 API 层获取锁方法时，会先尝试进行加锁操作（具体逻辑由实现类完成），如果加锁失败，会进入等待队列处理环节，这些处理逻辑同时也依赖最底层的基础数据提供层来完成。

3.1、原理概述

整个AQS实现线程同步的核心思想，可以用如下这段话来描述！

AQS 内部维护一个共享资源变量和线程等待队列，如果被请求的共享资源空闲，那么就将当前请求资源的线程设置为有效的工作线程，将共享资源设置为锁定状态；如果共享资源被占用，就需要一定的阻塞等待唤醒机制来保证锁分配。这个机制主要用的是 CLH 队列的变体实现的，将暂时获取不到锁的线程加入到等待队列中，待条件允许的时候将线程从队列中取出并进行唤醒。

CLH 队列是一个单向链表队列，对应的还有 CLH 锁实现，它是一个基于逻辑队列非线程饥饿的一种自旋公平锁实现，由 Craig、Landin 和 Hagersten 三位大佬发明，因此命名为 CLH 锁。关于这方面的技术知识讲解可以参阅这篇文章：多图详解 CLH 锁的原理与实现。

而AQS中的队列采用的是 CLH 变体的虚拟双向队列，通过将每一条请求共享资源的线程封装成一个 CLH 队列的一个节点来实现锁的分配。

具体实现原理，可以用如下图来简单概括：

图片图片

同时，AQS中维护了一个共享资源变量state，通过它来实现线程的同步状态控制，这个字段使用了volatile关键字修饰符来保证多线程下的可见性。

当多个线程尝试获取锁时，会通过CAS方式来修改state值，当state=1时表示当前对象锁已经被占有（相对独占模式来说），此时其他线程来加锁时会失败，加锁失败的线程会被放入上文说到的FIFO等待队列中，并且线程会被挂起，等待其他获取锁的线程释放锁才能够被唤醒。

总结下来，用大白话说就是，AQS是基于 CLH 队列，使用volatile修饰共享变量state，线程通过CAS方式去改变state状态值，如果成功则获取锁成功，失败则进入等待队列，等待被唤醒的线程同步器框架。

打开 ReentrantLock、ReadWriteLock、CountDownLatch、CyclicBarrier、Semaphore 等类的源码实现，你会发现它们的线程同步状态都是基于AQS实现的，可以看成是AQS的衍生物。

下面我们一起来看看相关的源码实现！

3.2、源码浅析

3.2.1、线程同步状态控制

AQS源码中维护的共享资源变量state，表示同步状态的意思，它是实现线程同步控制的关键字段，核心源码如下：

/**
 * The synchronization state.
 */
private volatile int state;

针对state字段值的获取和修改，AQS提供了三个方法，并且都采用Final修饰，意味着子类无法重写它们，相关方法如下：

方法

描述

protected final int getState()

获取state的值

protected final void setState(int newState)

设置state的值

protected final boolean compareAndSetState(int expect, int update)

使用 CAS 方式更新state

如果仔细分析源码，state字段还有一个很大的用处，通过它可以实现多线程的独占模式和共享模式。

以ReentrantLock和Semaphore类为例，它们的加锁过程中state值的变化情况如下。

3.2.1.1、ReentrantLock 独占模式的获取锁，简易流程图如下：

图片图片

ReentrantLock类部分核心源码，实现逻辑如下：

public class ReentrantLock implements Lock, java.io.Serializable {

    // 非公平锁实现类
    static final class NonfairSync extends Sync {
        private static final long serialVersionUID = 7316153563782823691L;

        // 加锁操作
        final void lock() {
            // 将state从0设置为1，如果成功，直接获取当前共享资源
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
                // 尝试加锁，会转调tryAcquire方法
                acquire(1);
        }

        protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            // 判断state是否等于0
            if (c == 0) {
                // 尝试state从0设置为1，如果成功，返回true
                if (compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            else if (current == getExclusiveOwnerThread()) {
                // 支持当前线程可重入，每调用一次，state的值加1
                int nextc = c + acquires;
                if (nextc < 0)
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }
    }
}
3.2.1.2、Semaphore 共享模式的获取锁，简易流程图如下：

图片图片

Semaphore类部分核心源码，实现逻辑如下：

public class Semaphore implements java.io.Serializable {

    // 初始化的时候，设置线程最大并发数，本质设置的是state的值
    public Semaphore(int permits) {
        sync = new NonfairSync(permits);
    }

    // 非公平锁内部实现类
    static final class NonfairSync extends Sync {

        NonfairSync(int permits) {
            // 设置state的值
            setState(permits);
        }

        // 通过共享方式，尝试获取锁
        protected int tryAcquireShared(int acquires) {
            return nonfairTryAcquireShared(acquires);
        }
    }

    // 尝试获取共享资源，会调用Sync.nonfairTryAcquireShared方法
    public boolean tryAcquire() {
        // 如果state的值小于0，表示无可用共享资源
        return sync.nonfairTryAcquireShared(1) >= 0;
    }

    // 抽象同步类
    abstract static class Sync extends AbstractQueuedSynchronizer {

        // 通过共享方式，尝试获取锁
        final int nonfairTryAcquireShared(int acquires) {
            for (;;) {
                // 通过cas方式，设置state自减
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }
    }
}
3.2.2、公平锁和非公平锁实现

在上文的ReentrantLock源码分析过程中，对于公平锁和非公平锁实现，其实已经有所解读。

在AQS中所有的加锁逻辑是有具有的实现类来完成，以ReentrantLock类为例，它的加锁逻辑由两个实现类来完成，分别是非公平锁静态内部实现类NonfairSync和公平锁静态内部实现类FairSync。

如上文的源码介绍，这两个类的的加锁逻辑基本一致，唯一的区别在于：公平锁实现类加锁时，增加了一个hasQueuedPredecessors()方法判断，这个方法会判断等待队列是否有线程处于等待状态，如果没有，尝试获取锁；如果有，就进入等待队列。

简单的说就是，非公平锁实现类的加锁方式，如果有线程尝试获取锁，直接尝试通过CAS方式进行抢锁，如果抢成功了，就直接获取锁，没有抢成功就进入等待队列；而公平锁实现类的加锁方式，会判断等待队列是否有线程处于等待状态，如果有则不去抢锁，乖乖排到后面，如果没有则尝试抢锁。

相对来说，非公平锁会有更好的性能，因为它的吞吐量比较大。其次，非公平锁让获取锁的时间变得更加不确定，可能会导致在阻塞队列中的线程长期处于饥饿状态。

Semaphore类的公平锁和非公平锁实现也类似，拥有两个静态内部实现类，源码就不再解读了，有兴趣的朋友可以自行阅读。

3.2.3、主要模板方法

从ReentrantLock的源码实现中可以看出，AQS使用了模板方法设计模式，它不提供加锁和释放锁的具体逻辑实现，而是由实现类重写对应的方法来完成，这样的好处就是实现更加的灵活，不同的线程同步器可以自行继承AQS类，然后实现独属于自身的加锁和解锁功能。

常用的模板方法主要有以下几个：

方法

描述

protected boolean isHeldExclusively()

判断该线程是否正在独占资源。只有用到Condition才需要去实现它

protected boolean tryAcquire(int arg)

独占方式。尝试获取资源，arg为获取锁的次数，成功则返回true，失败则返回false

protected boolean tryRelease(int arg)

独占方式。尝试释放资源，arg为释放锁的次数，成功则返回true，失败则返回false

protected int tryAcquireShared(int arg)

共享方式。尝试获取资源，arg为获取锁的次数，负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源

protected boolean tryReleaseShared(int arg)

共享方式。尝试释放资源，arg为释放锁的次数，如果释放后允许唤醒后续等待结点返回true，否则返回false

通常自定义线程同步器，要么是独占模式，要么是共享模式。

如果是独占模式，重写tryAcquire()和tryRelease()方法即可，比如 ReentrantLock 类。

如果是共享模式，重写tryAcquireShared()和tryReleaseShared()方法即可，比如 Semaphore 类。

3.2.4、线程加入等待队列实现

当线程调用tryAcquire()方法获取锁失败之后，就会调用addWaiter()方法，将当前线程加入到等待队列中去。

addWaiter()方法，部分核心源码如下：

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {

    // 将当前线程加入等待队列
    private Node addWaiter(Node mode) {
        // 以当前线程构造一个节点，尝试通过CAS方式插入到双向链表的队尾
        Node node = new Node(Thread.currentThread(), mode);
        Node pred = tail;
        if (pred != null) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        // 如果插入没有成功，则通过enq入队
        enq(node);
        return node;
    }

    // 通过enq入队
    private Node enq(final Node node) {
        // CAS 自旋方式，直到成功加入队尾
        for (;;) {
            Node t = tail;
            if (t == null) { 
                // 队列为空，创建一个空结点作为head结点，并将tail也指向它
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }
}

我们再来看看Node类节点相关的属性，部分核心源码如下：

static final class Node {

    // 当前节点在队列中的状态，状态值枚举含义如下：
    // 0:节点初始化时的状态
    // 1: 表示节点引用线程由于等待超时或被打断时的状态
    // -1: 表示当队列中加入后继节点被挂起时，其前驱节点会被设置为SIGNAL状态，表示该节点需要被唤醒
    // -2：当节点线程进入condition队列时的状态
    // -3：仅在释放共享锁releaseShared时对头节点使用
    volatile int waitStatus;

    // 前驱节点
    volatile Node prev;

    // 后继节点
    volatile Node next;

    //该节点的线程实例
    volatile Thread thread;

    // 指向下一个处于Condition状态的节点（用于条件队列）
    Node nextWaiter;

    //...
}

可以很清晰的看到，每个关键属性变量都加了volatile修饰符，确保多线程环境下可见。

正如上文所介绍的，Node其实是一个双向链表数据结构，大致的数据结构图如下！（图片来自ReentrantLock 的实现看 AQS 的原理及应用 - 美团技术团队）

图片图片

其中第一个节点，也叫头节点，为虚节点，并不存储任何线程信息，只是占位用；真正有数据的是从第二个节点开始，当有线程需要加入等待队列时，会向队尾进行插入。

线程加入等待队列之后，会再次调用acquireQueued()方法，尝试进行获取锁，如果成功或者中断就退出，部分核心源码如下：

final boolean acquireQueued(final Node node, int arg) {
    // 标记是否成功拿到锁
    boolean failed = true;
    try {
        // 标记等待过程中是否中断过
        boolean interrupted = false;
        // 开始自旋，要么获取锁，要么中断
        for (;;) {
            // 获取当前节点的前驱节点
            final Node p = node.predecessor();
            // 如果p是头结点，说明当前节点在等待队列的头部，尝试获取锁（头结点是虚节点）
            if (p == head && tryAcquire(arg)) {
                // 获取锁成功，头指针移动到当前node
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            // 如果p不是头节点或者是头节点但获取锁失败，判断当前节点是否要进入阻塞，如果满足要求，就通过park让线程进入阻塞状态，等待被唤醒
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            // 如果没有成功获取锁（比如超时或者被中断），那么取消节点在队列中的等待
            cancelAcquire(node);
    }
}

线程加入等待队列实现，总结下来，大致步骤如下：

  • 1.调用addWaiter()方法，将当前线程封装成一个节点，尝试通过CAS方式插入到双向链表的队尾，如果没有成功，再通过自旋方式插入，直到成功为止
  • 2.调用acquireQueued()方法，对在等待队列中排队的线程，尝试获取锁操作，如果失败，判断当前节点是否要进入阻塞，如果满足要求，就通过 LockSupport.park()方法让线程进入阻塞状态，并检查是否被中断，如果没有，等待被唤醒
3.2.5、线程从等待队列中被唤醒实现

当线程调用tryRelease()方法释放锁成功之后，会从等待队列的头部开始，获取排队的线程，并进行唤醒操作。

释放锁方法，部分核心源码如下：

public final boolean release(int arg) {
    // 尝试释放锁
    if (tryRelease(arg)) {
        // 获取头部节点
        Node h = head;
        if (h != null && h.waitStatus != 0)
            // 尝试唤醒头部节点的下一个节点中的线程
            unparkSuccessor(h);
        return true;
    }
    return false;
}

其中unparkSuccessor()是执行唤醒线程的核心方法，部分核心源码如下：

private void unparkSuccessor(Node node) {

    // 获取头结点 waitStatus

    int ws = node.waitStatus;

    // 置零当前线程所在的结点状态，允许失败

    if (ws  0) {

        s = null;

        // 从尾部节点开始向前找，找到队列中排在最前的有效节点

        for (Node t = tail; t != null && t != node; t = t.prev)

            if (t.waitStatus
LOVEHL^ˇ^
开发运维
0 0

相关文章

JavaScript2024新功能：Object.groupBy、正则表达式v标志
PHP trim 函数对多字节字符的使用和限制
新函数 json_validate() 、randomizer 类扩展…20 个PHP 8.3 新特性全面解析
使用HTMX为WordPress增效：如何在不使用复杂框架的情况下增强平台功能
为React 19做准备：WordPress 6.6用户指南
如何删除WordPress中的所有评论

发布评论

LOVEHL^ˇ^

这个人很懒，什么都没有留下～

文章8438 评论0