Java多线程之并发编程的核心AQS详解

2022-10-17,

AQS ( AbstractQueuedSynchronizer)是一个用来构建锁和同步器的框架,使用AQS能简单且高效地构造出应用广泛的大量的同步器,下面小编和大家来一起学习一下吧

目录
  • 一、AQS简介
    • 1.1、AOS概念
    •  1.2、AQS的核心思想
    • 1.3、AQS是自旋锁
    • 1.4、AQS支持两种资源分享的方式 
  • 二、AQS原理
    • 2.1、同步状态的管理
    • 2.2、等待队列
    • 2.3、CLH队列中的结点
    • 2.4、队列定义
    • 2.5、AQS底层的CAS机制
    • 2.6、通过ReentrantLock理解AQS
  • 三、AQS方法
    • 3.1、用户需要自己重写的方法
    • 3.2、AQS 提供的一系列模板方法
    • 3.3、acquire(int)方法
    • 3.4、release(int)方法
    • 3.5、acquireShared(int)方法
    • 3.6、releaseShared(int)方法
    • 参考链接:
  • 总结

    前言:Java并发包很多的同步工具类底层都是基于AQS来实现的,比如我们工作中经常用的Lock工具ReentrantLock、栅栏CountDownLatch、信号量Semaphore等。如果你想深入研究Java并发编程的话,那么AQS一定是绕不开的一块知识点,而且关于AQS的知识点也是面试中经常考察的内容,所以深入学习AQS很有必要。 

    学习AQS之前,我们有必要了解一下AQS底层中大量使用的CAS:Java多线程10:并发编程的基石CAS机制

    一、AQS简介

    1.1、AOS概念

    AQS,全名AbstractQueuedSynchronizer,是一个抽象类的队列式同步器,它的内部通过维护一个状态volatile int state(共享资源),一个FIFO线程等待队列来实现同步功能。AQS类是整个 JUC包的核心类,JUC 中的ReentrantLock、ReentrantReadWriteLock、CountDownLatch、Semaphore和LimitLatch等同步工具都是基于AQS实现的。

    state用关键字volatile修饰,代表着该共享资源的状态一更改就能被所有线程可见,而AQS的加锁方式本质上就是多个线程在竞争state,当state为0时代表线程可以竞争锁,不为0时代表当前对象锁已经被占有,其他线程来加锁时则会失败,加锁失败的线程会被放入一个FIFO的等待队列中,这些线程会被UNSAFE.park()操作挂起,等待其他获取锁的线程释放锁才能够被唤醒。

    而这个等待队列其实就相当于一个CLH队列,用一张原理图来表示大致如下:

     1.2、AQS的核心思想

    如果被请求的共享资源空闲,则将当前请求资源的线程设置为有效的工作线程,并将共享资源设置为锁定状态,如果被请求的共享资源被占用,那么就需要一套线程阻塞等待以及被唤醒时锁分配的机制,这个机制AQS是用CLH队列锁实现的,即将暂时获取不到锁的线程加入到队列中。CLH(Craig,Landin,and Hagersten)队列是一个虚拟的双向队列,虚拟的双向队列即不存在队列实例,仅存在节点之间的关联关系。
    AQS是将每一条请求共享资源的线程封装成一个CLH锁队列的一个结点(Node),来实现锁的分配。

    用大白话来说,AQS就是基于CLH队列,用volatile修饰共享变量state,线程通过CAS去改变状态符,成功则获取锁成功,失败则进入等待队列,等待被唤醒。

    1.3、AQS是自旋锁

    AQS是自旋锁:在等待唤醒的时候,经常会使用自旋(while(!cas()))的方式,不停地尝试获取锁,直到被其他线程获取成功

    实现了AQS的锁有:自旋锁、互斥锁、读锁写锁、条件产量、信号量、栅栏都是AQS的衍生物

    1.4、AQS支持两种资源分享的方式 

    Exclusive(独占,只有一个线程能执行,如ReentrantLock)和Share(共享,多个线程可同时执行,如Semaphore/CountDownLatch)。

    自定义的同步器继承AQS后,只需要实现共享资源state的获取和释放方式即可,其他如线程队列的维护(如获取资源失败入队/唤醒出队等)等操作,AQS在底层已经实现了。

    线程的阻塞和唤醒

    在JDK1.5之前,除了内置的监视器机制外,没有其它方法可以安全且便捷得阻塞和唤醒当前线程。

    JDK1.5以后,java.util.concurrent.locks包提供了LockSupport类来作为线程阻塞和唤醒的工具。

    二、AQS原理

    2.1、同步状态的管理

    同步状态,其实就是资源。AQS使用单个int(32位)来保存同步状态,并暴露出getState、setState以及compareAndSetState操作来读取和更新这个状态。

    public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer
        implements java.io.Serializable {
     
        private volatile int state;
      
        protected final int getState() {
             return state;
        }
     
        protected final void setState(int newState) {
           state = newState;
        }
     
        protected final boolean compareAndSetState(int expect, int update) {
         // See below for intrinsics setup to support this
         return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
        }
     
        //省略展示其它代码...
    }

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

     

    2.2、等待队列

    等待队列,是AQS框架的核心,整个框架的关键其实就是如何在并发状态下管理被阻塞的线程。

    等待队列是严格的FIFO队列,是Craig,Landin和Hagersten锁(CLH锁)的一种变种,采用双向循环链表实现,因此也叫CLH队列。

    2.3、CLH队列中的结点

    AQS内部还定义了一个静态类Node,表示CLH队列的每一个结点,该结点的作用是对每一个等待获取资源做了封装,包含了需要同步的线程本身、线程等待状态....

    LH队列中的结点是对线程的包装,结点一共有两种类型:独占(EXCLUSIVE)和共享(SHARED)。

    每种类型的结点都有一些状态,其中独占结点使用其中的CANCELLED(1)、SIGNAL(-1)、CONDITION(-2),共享结点使用其中的CANCELLED(1)、SIGNAL(-1)、PROPAGATE(-3)。

    结点状态 描述
    CANCELLED 1 取消。表示后驱结点被中断或超时,需要移出队列
    SIGNAL -1 发信号。表示后驱结点被阻塞了(当前结点在入队后、阻塞前,应确保将其prev结点类型改为SIGNAL,以便prev结点取消或释放时将当前结点唤醒。)
    CONDITION -2 Condition专用。表示当前结点在Condition队列中,因为等待某个条件而被阻塞了
    PROPAGATE -3 传播。适用于共享模式(比如连续的读操作结点可以依次进入临界区,设为PROPAGATE有助于实现这种迭代操作。)
    INITIAL 0 默认。新结点会处于这种状态

    2.4、队列定义

    对于CLH队列,当线程请求资源时,如果请求不到,会将线程包装成结点,将其挂载在队列尾部。

    下面结合代码一起看下节点进入队列的过程。

       private Node enq(final Node node) {
            for (;;) {
                Node t = tail;   // 1
                if (t == null) { // Must initialize
                    if (compareAndSetHead(new Node()))  // 2 
                        tail = head;
                } else {
                    node.prev = t; // 3
                    if (compareAndSetTail(t, node)) { // 4
                        t.next = node;
                        return t;
                    }
                }
            }
        }

    2.5、AQS底层的CAS机制

    在研究JDK中AQS时,会发现这个类很多地方都使用了CAS操作,在并发实现中CAS操作必须具备原子性,而且是硬件级别的原子性,Java被隔离在硬件之上,明显力不从心,这时为了能直接操作操作系统层面,肯定要通过用C++编写的native本地方法来扩展实现。JDK提供了一个类来满足CAS的要求,sun.misc.Unsafe,从名字上可以大概知道它用于执行低级别、不安全的操作,AQS就是使用此类完成硬件级别的原子操作。UnSafe通过JNI调用本地C++代码,C++代码调用CPU硬件指令集。

    Unsafe是一个很强大的类,它可以分配内存、释放内存、可以定位对象某字段的位置、可以修改对象的字段值、可以使线程挂起、使线程恢复、可进行硬件级别原子的CAS操作等等。

    2.6、通过ReentrantLock理解AQS

    ReentrantLock中公平锁和非公平锁在底层是相同的,这里以非公平锁为例进行分析。

    在非公平锁中,有一段这样的代码:

    // java.util.concurrent.locks.ReentrantLock
     
    static final class NonfairSync extends Sync {
        ...
        final void lock() {
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
                acquire(1);
        }
      ...
    }

    看一下这个Acquire是怎么写的:

    // java.util.concurrent.locks.AbstractQueuedSynchronizer
     
    public final void acquire(int arg) {
        if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

    再看一下tryAcquire方法:

    // java.util.concurrent.locks.AbstractQueuedSynchronizer
     
    protected boolean tryAcquire(int arg) {
        throw new UnsupportedOperationException();
    }

    可以看出,这里只是AQS的简单实现,具体获取锁的实现方法是由各自的公平锁和非公平锁单独实现的(以ReentrantLock为例)。如果该方法返回了True,则说明当前线程获取锁成功,就不用往后执行了;如果获取失败,就需要加入到等待队列中。

    三、AQS方法

    AQS代码内部提供了一系列操作锁和线程队列的方法,主要操作锁的方法包含以下几个:

    compareAndSetState():利用CAS的操作来设置state的值

    tryAcquire(int):独占方式获取锁。成功则返回true,失败则返回false。

    tryRelease(int):独占方式释放锁。成功则返回true,失败则返回false。

    tryReleaseShared(int):共享方式释放锁。如果释放后允许唤醒后续等待结点返回true,否则返回false。

    像ReentrantLock就是实现了自定义的tryAcquire-tryRelease,从而操作state的值来实现同步效果。

    3.1、用户需要自己重写的方法

    上面介绍到 AQS 已经帮用户解决了同步器定义过程中的大部分问题,只将下面两个问题丢给用户解决:

    • 什么是资源
    • 什么情况下资源是可以被访问的

    具体的,AQS 是通过暴露以下 API 来让用户解决上面的问题的。

    钩子方法 描述
    tryAcquire 独占方式。尝试获取资源,成功则返回true,失败则返回false。
    tryRelease 独占方式。尝试释放资源,成功则返回true,失败则返回false。
    tryAcquireShared 共享方式。尝试获取资源。负数表示失败;0表示成功,但没有剩余可用资源;正数表示成功,且有剩余资源。
    tryReleaseShared 共享方式。尝试释放资源,如果释放后允许唤醒后续等待结点返回true,否则返回false。
    isHeldExclusively 该线程是否正在独占资源。只有用到condition才需要去实现它。

    如果你需要实现一个自己的同步器,一般情况下只要继承 AQS ,并重写 AQS 中的这个几个方法就行了。至于具体线程等待队列的维护(如获取资源失败入队/唤醒出队等),AQS已经在顶层实现好了。要不怎么说Doug Lea贴心呢。

    需要注意的是:如果你没在子类中重写这几个方法就直接调用了,会直接抛出异常。所以,在你调用这些方法之前必须重写他们。不使用的话可以不重写。

    3.2、AQS 提供的一系列模板方法

    查看 AQS 的源码我们就可以发现这个类提供了很多方法,看起来让人“眼花缭乱”的。但是最主要的两类方法就是获取资源的方法和释放资源的方法。因此我们抓住主要矛盾就行了:

    • public final void acquire(int arg) // 独占模式的获取资源
    • public final boolean release(int arg) // 独占模式的释放资源
    • public final void acquireShared(int arg) // 共享模式的获取资源
    • public final boolean releaseShared(int arg) // 共享模式的释放资源

    3.3、acquire(int)方法

    该方法以独占方式获取资源,如果获取到资源,线程继续往下执行,否则进入等待队列,直到获取到资源为止,且整个过程忽略中断的影响。该方法是独占模式下线程获取共享资源的顶层入口。

    public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

    下面分析下这个acquire方法的具体执行流程:

    step1:首先这个方法调用了用户自己实现的方法tryAcquire方法尝试获取资源,如果这个方法返回true,也就是表示获取资源成功,那么整个acquire方法就执行结束了,线程继续往下执行;

    step2:如果tryAcquir方法返回false,也就表示尝试获取资源失败。这时acquire方法会先调用addWaiter方法将当前线程封装成Node类并加入一个FIFO的双向队列的尾部。

    step3:再看acquireQueued这个关键方法。首先要注意的是这个方法中哪个无条件的for循环,这个for循环说明acquireQueued方法一直在自旋尝试获取资源。进入for循环后,首先判断了当前节点的前继节点是不是头节点,如果是的话就再次尝试获取资源,获取资源成功的话就直接返回false(表示未被中断过)

    假如还是没有获取资源成功,判断是否需要让当前节点进入waiting状态,经过 shouldParkAfterFailedAcquire这个方法判断,如果需要让线程进入waiting状态的话,就调用LockSupport的park方法让线程进入waiting状态。进入waiting状态后,这线程等待被interupt或者unpark(在release操作中会进行这样的操作,可以参见后面的代码)。这个线程被唤醒后继续执行for循环来尝试获取资源。

    final boolean acquireQueued(final Node node, int arg) {
            boolean failed = true;
            try {
                boolean interrupted = false;
                for (;;) {
                    final Node p = node.predecessor();
                    //首先判断了当前节点的前继节点是不是头节点,如果是的话就再次尝试获取资源,
                    //获取资源成功的话就直接返回false(表示未被中断过)
                    if (p == head && tryAcquire(arg)) {
                        setHead(node);
                        p.next = null; // help GC
                        failed = false;
                        return interrupted;
                    }
                    //判断是否需要让当前节点进入waiting状态
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        // 如果在整个等待过程中被中断过,则返回true,否则返回false。
                        // 如果线程在等待过程中被中断过,它是不响应的。只是获取资源后才再进行自我中断selfInterrupt(),将中断补上。
                        interrupted = true;
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }

    以上就是acquire方法的简单分析。

    单独看这个方法的话可能会不太清晰,结合ReentrantLock、ReentrantReadWriteLock、CountDownLatch、Semaphore和LimitLatch等同步工具看这个代码的话就会好理解很多。

    3.4、release(int)方法

    release(int)方法是独占模式下线程释放共享资源的顶层入口。它会释放指定量的资源,如果彻底释放了(即state=0),它会唤醒等待队列里的其他线程来获取资源。

    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();
    }
     
    private void unparkSuccessor(Node node) {
        /*
             * If status is negative (i.e., possibly needing signal) try
             * to clear in anticipation of signalling.  It is OK if this
             * fails or if status is changed by waiting thread.
             */
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);
     
        /*
             * Thread to unpark is held in successor, which is normally
             * just the next node.  But if cancelled or apparently null,
             * traverse backwards from tail to find the actual
             * non-cancelled successor.
             */
        Node s = node.next;
        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;
        }
        if (s != null)
            LockSupport.unpark(s.thread);
    }

    与acquire()方法中的tryAcquire()类似,tryRelease()方法也是需要独占模式的自定义同步器去实现的。正常来说,tryRelease()都会成功的,因为这是独占模式,该线程来释放资源,那么它肯定已经拿到独占资源了,直接减掉相应量的资源即可(state-=arg),也不需要考虑线程安全的问题。

    但要注意它的返回值,上面已经提到了,release()是根据tryRelease()的返回值来判断该线程是否已经完成释放掉资源了!所以自义定同步器在实现时,如果已经彻底释放资源(state=0),要返回true,否则返回false。

    unparkSuccessor(Node)方法用于唤醒等待队列中下一个线程。这里要注意的是,下一个线程并不一定是当前节点的next节点,而是下一个可以用来唤醒的线程,如果这个节点存在,调用unpark()方法唤醒。

    总之,release()是独占模式下线程释放共享资源的顶层入口。它会释放指定量的资源,如果彻底释放了(即state=0),它会唤醒等待队列里的其他线程来获取资源。(需要注意的是队列中被唤醒的线程不一定能立马获取资源,因为资源在释放后可能立马被其他线程(不是在队列中等待的线程)抢掉了

    3.5、acquireShared(int)方法

    acquireShared(int)方法是共享模式下线程获取共享资源的顶层入口。它会获取指定量的资源,获取成功则直接返回,获取失败则进入等待队列,直到获取到资源为止,整个过程忽略中断。

    public final void acquireShared(int arg) {
        //tryAcquireShared需要用户自定义实现
        if (tryAcquireShared(arg) < 0)
            doAcquireShared(arg);
    }

    可以发现,这个方法的关键实现其实是获取资源失败后,怎么管理线程。也就是doAcquireShared的逻辑。

    //不响应中断
    private void doAcquireShared(int arg) {
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                if (p == head) {
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        if (interrupted)
                            selfInterrupt();
                        failed = false;
                        return;
                    }
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

    可以看出,doAcquireShared的逻辑和acquireQueued的逻辑差不多。将当前线程加入等待队列尾部休息,直到其他线程释放资源唤醒自己,自己成功拿到相应量的资源后才返回。

    简单总结下acquireShared的流程:

    step1:tryAcquireShared()尝试获取资源,成功则直接返回;

    step2:失败则通过doAcquireShared()进入等待队列park(),直到被unpark()/interrupt()并成功获取到资源才返回。整个等待过程也是忽略中断的。

    3.6、releaseShared(int)方法

    releaseShared(int)方法是共享模式下线程释放共享资源的顶层入口。它会释放指定量的资源,如果成功释放且允许唤醒等待线程,它会唤醒等待队列里的其他线程来获取资源。

    public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }

    释放掉资源后,唤醒后继。跟独占模式下的release()相似,但有一点稍微需要注意:独占模式下的tryRelease()在完全释放掉资源(state=0)后,才会返回true去唤醒其他线程,这主要是基于独占下可重入的考量;而共享模式下的releaseShared()则没有这种要求,共享模式实质就是控制一定量的线程并发执行,那么拥有资源的线程在释放掉部分资源时就可以唤醒后继等待结点。

    参考链接:

    从ReentrantLock的实现看AQS的原理及应用

    并发编程的基石——AQS类

    总结

    本篇文章就到这里了,希望能够给你带来帮助,也希望您能够多多关注北冥有鱼的更多内容!

    《Java多线程之并发编程的核心AQS详解.doc》

    下载本文的Word格式文档,以方便收藏与打印。