理解AQS的原理及应用总结

理解AQS的原理及应用总结一 AQS 基本概述 AQS 全称为 AbstractQueu 是 Java 中用于构建锁和同步器的框架性组件 它是 Java 并发包中 ReentrantLoc Semaphore ReentrantRea 等同步器的基础 设计思想 AQS 的设计思想是 在其内部维护了一个双向队列

大家好,我是讯享网,很高兴认识大家。

一、AQS基本概述

AQS全称为AbstractQueuedSynchronizer,是Java中用于构建锁和同步器的框架性组件,它是Java并发包中ReentrantLock、Semaphore、ReentrantReadWriteLock等同步器的基础。

设计思想

AQS的设计思想是,在其内部维护了一个双向队列,用于管理请求锁的线程。当有线程请求锁时,AQS会将其封装成一个Node节点,并加入到等待队列中,线程则会进入阻塞状态。当持有锁的线程释放锁时,AQS会从等待队列中唤醒一个线程来获取锁,从而实现线程的同步和互斥。

主要特点

AQS的主要特点包括:

  1. 支持独占模式和共享模式。独占模式下只允许一个线程持有锁,共享模式下可以允许多个线程同时持有锁。
  2. 内部维护了一个双向队列,用于管理请求锁的线程,队列中的节点是线程的封装。
  3. 通过CAS(Compare And Swap)操作实现状态的改变,状态可以是任意int类型的变量。
  4. 具有可重入性,即同一个线程可以多次获取同一把锁而不会出现死锁。

AQS的实现被广泛应用于Java并发包中的各种同步器,如ReentrantLock、ReentrantReadWriteLock、Semaphore、CountDownLatch等。AQS为这些同步器提供了一个统一的基础框架,并且可以让开发人员基于此进行扩展和定制化。

总之,AQS提供了一种高效且灵活的实现同步器的方式,为Java中的并发编程提供了基础设施。通过使用AQS,开发人员可以避免自己重复实现同步器的底层机制,从而更加专注于业务的实现。

二、AQS前置知识点

2.1 设计模式——模板方法

AbstractQueuedSynchronizer是个抽象类,所有用到方法的类都要继承此类的若干方法,对应的设计模式就是模版模式

模版模式定义:一个抽象类公开定义了执行它的方法的方式/模板。它的子类可以按需要重写方法实现,但调用将以抽象类中定义的方式进行。这种类型的设计模式属于行为型模式。

2.2 LookSupport

LockSupport 是一个线程阻塞工具类,所有的方法都是静态方法,可以让线程在任意位置阻塞,当然阻塞之后肯定得有唤醒的方法。常用方法如下:

public static void park(Object blocker); // 暂停当前线程 public static void parkNanos(Object blocker, long nanos); // 暂停当前线程,不过有超时时间的限制 public static void parkUntil(Object blocker, long deadline); // 暂停当前线程,直到某个时间 public static void park(); // 无期限暂停当前线程 public static void parkNanos(long nanos); // 暂停当前线程,不过有超时时间的限制 public static void parkUntil(long deadline); // 暂停当前线程,直到某个时间 public static void unpark(Thread thread); // 恢复当前线程 public static Object getBlocker(Thread t); 

讯享网

park是因为park英文意思为停车。我们如果把Thread看成一辆车的话,park就是让车停下,unpark就是让车启动然后跑起来。

与Object类的wait/notify机制相比,park/unpark有两个优点:

  1. thread为操作对象更符合阻塞线程的直观定义
  2. 操作更精准,可以准确地唤醒某一个线程(notify随机唤醒一个线程,notifyAll 唤醒所有等待的线程),增加了灵活性。

park/unpark调用的是 Unsafe(提供CAS操作) 中的 native代码。

2.3 CAS

CAS 是 CPU指令级别实现了原子性的比较和交换(Conmpare And Swap)操作,注意CAS不是锁只是CPU提供的一个原子性操作指令。


讯享网

CAS在语言层面不进行任何处理,直接将原则操作实现在硬件级别实现,之所以可以实现硬件级别的操作核心是因为CAS操作类中有个核心类UnSafe类。

关于CAS引发的ABA问题、性能开销问题、只能保证一个共享变量之间的原则性操作问题,以前CAS中写过,在此不再重复讲解。

2.4 条件变量

Object的wait、notify函数是配合Synchronized锁实现线程间同步协作的功能,AQS的ConditionObject条件变量也提供这样的功能,通过ConditionObject的await和signal两类函数完成。

不同于Synchronized锁,一个AQS可以对应多个条件变量,而Synchronized只有一个。

如上图所示,ConditionObject内部维护着一个单向条件队列,不同于CHL队列,条件队列只入队执行await的线程节点,并且加入条件队列的节点,不能在CHL队列, 条件队列出队的节点,会入队到CHL队列。

当某个线程执行了ConditionObject的await函数,阻塞当前线程,线程会被封装成Node节点添加到条件队列的末端,其他线程执行ConditionObject的signal函数,会将条件队列头部线程节点转移到CHL队列参与竞争资源,具体流程如下图

三、AQS应用介绍

3.1 开发中的基本应用指导

在开发中,我们可以利用AQS提供的同步机制来实现线程的协作和同步,从而达到线程安全的目的。以下是一些常见的开发应用场景:

  1. 使用ReentrantLock实现同步:ReentrantLock是一个可重入独占锁,可以使用它来实现线程的同步。
  2. 使用ReentrantReadWriteLock实现读写锁:ReentrantReadWriteLock是一个可重入的读写锁,可以使用它来实现对共享资源的读写操作。
  3. 使用Semaphore实现信号量:Semaphore是一种计数信号量,可以用来控制同时访问某个共享资源的线程个数。
  4. 使用CountDownLatch实现线程的等待和唤醒:CountDownLatch可以让某个线程等待其他线程执行完毕后再继续执行。
  5. 使用CyclicBarrier实现线程的协作:CyclicBarrier可以让一组线程相互等待,直到所有线程都到达某个状态后才会继续执行。
  6. 使用Condition实现线程的等待和唤醒:Condition是一种条件变量,可以让线程在某个条件满足时等待,直到另一个线程发出唤醒信号后才继续执行。
  7. ThreadPoolExecutor:Worker利用AQS同步状态实现对独占线程变量的设置(tryAcquire和tryRelease)。

除此之外,AQS还可以用来实现自定义的同步器,比如读写锁、可重入锁等。总之,在开发中,我们可以根据具体的业务需求和线程安全要求,灵活地使用AQS提供的各种同步机制。

3.2 框架中的应用展示举例

ReentrantLock的可重入性是AQS很好的应用之一,在ReentrantLock里面,不管是公平锁还是非公平锁,都有下面一段逻辑

讯享网// java.util.concurrent.locks.ReentrantLock.FairSync#tryAcquire if (c == 0) { if (!hasQueuedPredecessors() && compareAndSetState(0, acquires)) { setExclusiveOwnerThread(current); return true; } } else if (current == getExclusiveOwnerThread()) { int nextc = c + acquires; if (nextc < 0) throw new Error("Maximum lock count exceeded"); setState(nextc); return true; } // java.util.concurrent.locks.ReentrantLock.Sync#nonfairTryAcquire if (c == 0) { if (compareAndSetState(0, acquires)){ setExclusiveOwnerThread(current); 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; }

可以看到,有一个同步状态State来控制整体可重入的情况。State是Volatile修饰的,用于保证一定的可见性和有序性。

// java.util.concurrent.locks.AbstractQueuedSynchronizer private volatile int state;
  1. State初始化的时候为0,表示没有任何线程持有锁。
  2. 当有线程持有该锁时,值就会在原来的基础上+1,同一个线程多次获得锁时,就会多次+1,这里就是可重入的概念。
  3. 解锁也是对这个字段-1,一直到0,此线程对锁释放。
3.3 开发应用举例
自定义同步工具

实现一个同步工具基本代码如下:

讯享网package org.zyf.javabasic.thread.toolstest.zyf; import java.util.concurrent.locks.AbstractQueuedSynchronizer; public class ZYFLock { private static class Sync extends AbstractQueuedSynchronizer { @Override protected boolean tryAcquire(int arg) { return compareAndSetState(0, 1); } @Override protected boolean tryRelease(int arg) { setState(0); return true; } @Override protected boolean isHeldExclusively() { return getState() == 1; } } private Sync sync = new Sync(); public void lock() { sync.acquire(1); } public void unlock() { sync.release(1); } }
package org.zyf.javabasic.thread.toolstest.zyf; / * @author yanfengzhang * @description * @date 2020/5/2 10:50 */ public class ZYFLockTest { static int count = 0; static ZYFLock zyfLock = new ZYFLock(); public static void main(String[] args) throws InterruptedException { Runnable runnable = new Runnable() { @Override public void run() { try { zyfLock.lock(); for (int i = 0; i < 10000; i++) { count++; } } catch (Exception e) { e.printStackTrace(); } finally { zyfLock.unlock(); } } }; Thread thread1 = new Thread(runnable); Thread thread2 = new Thread(runnable); thread1.start(); thread2.start(); thread1.join(); thread2.join(); //输入结果为20000 System.out.println(count); } }
3.4 AQS 对资源的共享方式

AQS定义两种资源共享方式:

【1】Exclusive(独占):只有一个线程能执行,如 ReentrantLock。又可分为公平锁和非公平锁:
● 公平锁:按照线程在队列中的排队顺序,先到者先拿到锁;
● 非公平锁:当线程要获取锁时,无视队列顺序直接去抢锁,谁抢到就是谁的;

【2】Share(共享):多个线程可同时执行,如Semaphore/CountDownLatch。

ReentrantReadWriteLock 可以看成是组合式,因为 ReentrantReadWriteLock也就是读写锁允许多个线程对某资源进行读。

不同的自定义同步器争用共享资源的方式也不同。自定义同步器在实现时只需要实现共享资源 state 的获取与释放方式即可,至于具体线程等待队列的维护(如获取资源失败入队/唤醒出队等),AQS已经在上层已经帮我们实现好了。


四、ReentrantLock基本知识与AQS的联系

4.1 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(); } }

4.2与AQS联系---Acquire方法

ReentrantLock支持公平锁和非公平锁,并且ReentrantLock的底层就是由AQS来实现的,着重从这两者的加锁过程来理解与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); } ... } // java.util.concurrent.locks.ReentrantLock#FairSync static final class FairSync extends Sync { ... final void lock() { acquire(1); } ... }

这块代码的含义的重点在于为获取锁失败,则进入Acquire方法进行后续处理。结合公平锁和非公平锁的加锁流程,虽然流程上有一定的不同,但是都调用了Acquire方法,而Acquire方法是FairSync和UnfairSync的父类AQS中的核心方法。

五、AQS原理分析

5.1 原理概览
架构图分析


通过下面的架构图来整体了解一下AQS框架,有颜色的为Method,无颜色的为Attribution,总的来说,AQS框架共分为五层,自上而下由浅入深,从AQS对外暴露的API到底层基础数据。

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

AQS中队列:CLH变体的虚拟双向队列(FIFO)

AQS核心思想是,如果被请求的共享资源空闲,那么就将当前请求资源的线程设置为有效的工作线程,将共享资源设置为锁定状态;如果共享资源被占用,就需要一定的阻塞等待唤醒机制来保证锁分配。这个机制主要用的是CLH队列(CLH:Craig、Landin and Hagersten队列,是单向链表)的变体实现的,将暂时获取不到锁的线程加入到队列中(AQS是通过将每条请求共享资源的线程封装成一个节点来实现锁的分配)

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

或者如下图

head 头结点又叫哨兵节点,线程thread为null

 AQS数据结构

AbstractQueuedSynchronizer 类底层的数据结构是使用 CLH(Craig,Landin,and Hagersten) 队列是一个虚拟的双向队列。其中Sync queue,即同步队列,是双向链表,包括 head结点和 tail结点,head结点主要用作后续的调度。而 Condition queue不是必须的,其是一个单向链表,只有当使用 Condition时,才会存在此单向链表。并且可能会有多个 Condition queue。

Condition(条件队列,又叫等待队列)

Condition是Java并发API中的一个特性,它通常与ReentrantLock结合使用,用于实现更灵活的线程等待和通知机制。它提供了类似于Object的wait()和notify()方法的功能,但比传统的wait()和notify()更加灵活和安全。

主要方法:
- await():使当前线程等待,直到另一个线程调用相应的signal()或signalAll()方法唤醒它。
- awaitUninterruptibly():与await()类似,但不会响应中断。
- signal():唤醒等待在该条件队列上的一个线程。
- signalAll():唤醒等待在该条件队列上的所有线程。
 

总的来说,Condition的本质就是等待队列和同步队列的交互

当一个持有锁的线程调用Condition.await()时,它会执行以下步骤:

1.构造一个新的等待队列节点加入到等待队列队尾
2.释放锁,也就是将它的同步队列节点从同步队列队首移除
3.自旋,直到它在等待队列上的节点移动到了同步队列(通过其他线程调用signal())或被中断
4.阻塞当前节点,直到它获取到了锁,也就是它在同步队列上的节点排队排到了队首。

当一个持有锁的线程调用Condition.signal()时,它会执行以下操作:

从等待队列的队首开始,尝试对队首节点执行唤醒操作;如果CANCELLED,就尝试唤醒下一个节点;如果再CANCELLED则继续迭代。

对每个节点执行唤醒操作时,首先将节点加入同步队列,此时await()操作的步骤3的解锁条件就已经开启了。然后分两种情况讨论:

主要特点:
- 插入和移除操作必须同时发生,否则线程将被阻塞。
- 典型的用途是实现线程之间的直接传输,用于传递数据或任务。

SynchronousQueue可以被用于一些线程之间的交互场景,其中生产者线程将数据传递给消费者线程,并且这种传递必须是同时发生的。

Condition`用于在特定条件下挂起和唤醒线程,通常与`ReentrantLock`一起使用,而`SynchronousQueue`用于实现线程之间的直接传输,通过插入和移除操作的同步来实现数据或任务的传递。

Node


 AQS中最基本的数据结构——Node(即为上面CLH变体队列中的节点),其基本源码如下:

讯享网static final class Node { // 标识节点当前在共享模式下 static final Node SHARED = new Node(); // 标识节点当前在独占模式下 static final Node EXCLUSIVE = null; // ======== 下面的几个int常量是给waitStatus用的 =========== / waitStatus value to indicate thread has cancelled */ // 代码此线程取消了争抢这个锁 static final int CANCELLED = 1; / waitStatus value to indicate successor's thread needs unparking */ // 官方的描述是,其表示当前node的后继节点对应的线程需要被唤醒 static final int SIGNAL = -1; / waitStatus value to indicate thread is waiting on condition */ // 本文不分析condition,所以略过吧,下一篇文章会介绍这个 static final int CONDITION = -2; / * waitStatus value to indicate the next acquireShared should * unconditionally propagate */ // 同样的不分析,略过吧 static final int PROPAGATE = -3; // ===================================================== // 取值为上面的1、-1、-2、-3,或者0(以后会讲到) // 这么理解,暂时只需要知道如果这个值 大于0 代表此线程取消了等待, // ps: 半天抢不到锁,不抢了,ReentrantLock是可以指定timeouot的。。。 volatile int waitStatus; // 前驱节点的引用 volatile Node prev; // 后继节点的引用 volatile Node next; // 这个就是线程本尊 volatile Thread thread; //指向下一个处于CONDITION状态的节点 Node nextWaiter; //返回前驱节点,没有的话抛出npe final Node predecessor() throws NullPointerException { Node p = prev; if (p == null) throw new NullPointerException(); else return p; } }

其中waitStatus有下面几个枚举值如下:

nextWaiter特殊标记

  • NodeCLH队列时,nextWaiter表示共享式或独占式标记
  • Node在条件队列时,nextWaiter表示下个Node节点指针

同步状态State

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

// java.util.concurrent.locks.AbstractQueuedSynchronizer private volatile int state;
讯享网//获取State的值 protected final int getState() //设置State的值 protected final void setState(int newState) //使用CAS方式更新State protected final boolean compareAndSetState(int expect, int update)

这几个方法都是final修饰的,说明子类中无法重写它们。我们可以通过修改State字段表示的同步状态来实现多线程的独占模式和共享模式(加锁过程)(对于我们自定义的同步工具,需要自定义获取同步状态和释放状态的方式)。

5.2 AQS重要方法(以ReentrantLock关联分析)

从架构图中可以得知,AQS提供了大量用于自定义同步器实现的Protected方法。自定义同步器实现的相关方法也只是为了通过修改State字段来实现多线程的独占模式或者共享模式。部分展示如下:

一般来说,自定义同步器要么是独占方式,要么是共享方式,它们也只需实现tryAcquire-tryRelease、tryAcquireShared-tryReleaseShared中的一种即可。AQS也支持自定义同步器同时实现独占和共享两种方式,如ReentrantReadWriteLock。ReentrantLock是独占锁,所以实现了tryAcquire-tryRelease。

以非公平锁为例,这里主要阐述一下非公平锁与AQS之间方法的关联之处,基本流程图如下:

以非公平锁为例,总结如下:

5.3 通过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); } ... } // java.util.concurrent.locks.AbstractQueuedSynchronizer public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); } // java.util.concurrent.locks.AbstractQueuedSynchronizer protected boolean tryAcquire(int arg) { throw new UnsupportedOperationException(); }
线程加入等待队列


当执行Acquire(1)时,会通过tryAcquire获取锁。在这种情况下,如果获取锁失败,就会调用addWaiter加入到等待队列中去。获取锁失败后,会执行addWaiter(Node.EXCLUSIVE)加入等待队列,具体实现方法如下:

讯享网// java.util.concurrent.locks.AbstractQueuedSynchronizer 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(node); return node; } private final boolean compareAndSetTail(Node expect, Node update) { return unsafe.compareAndSwapObject(this, tailOffset, expect, update); } // java.util.concurrent.locks.AbstractQueuedSynchronizer static { try { stateOffset = unsafe.objectFieldOffset(AbstractQueuedSynchronizer.class.getDeclaredField("state")); headOffset = unsafe.objectFieldOffset(AbstractQueuedSynchronizer.class.getDeclaredField("head")); tailOffset = unsafe.objectFieldOffset(AbstractQueuedSynchronizer.class.getDeclaredField("tail")); waitStatusOffset = unsafe.objectFieldOffset(Node.class.getDeclaredField("waitStatus")); nextOffset = unsafe.objectFieldOffset(Node.class.getDeclaredField("next")); } catch (Exception ex) { throw new Error(ex); } } // java.util.concurrent.locks.AbstractQueuedSynchronizer 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; } } } }
  • 通过当前的线程和锁模式新建一个节点。
  • Pred指针指向尾节点Tail。
  • 将New中Node的Prev指针指向Pred。
  • 通过compareAndSetTail方法,完成尾节点的设置。这个方法主要是对tailOffset和Expect进行比较,如果tailOffset的Node和Expect的Node地址相同,那么设置Tail的值为Update的值。
  • 如果Pred指针是Null(说明等待队列中没有元素),或者当前Pred指针和Tail指向的位置不同(说明被别的线程已经修改),就需要看一下Enq的方法。
  • 如果没有被初始化,需要进行初始化一个头结点出来。但请注意,初始化的头结点并不是当前线程节点,而是调用了无参构造函数的节点。如果经历了初始化或者并发导致队列中有元素,则与之前的方法相同。其实,addWaiter就是一个在双端链表添加尾节点的操作,需要注意的是,双端链表的头结点是一个无参构造函数的头结点。
等待队列中线程出队列时机

一个线程获取锁失败了,被放入等待队列,acquireQueued会把放入队列中的线程不断去获取锁,直到获取成功或者不再需要获取(中断)。

从“何时出队列”和“如何出队?”两个方向来分析一下acquireQueued源码:

// java.util.concurrent.locks.AbstractQueuedSynchronizer 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为头节点且当前没有获取到锁(可能是非公平锁被抢占了)或者是p不为头结点,这个时候就要判断当前node是否要被阻塞(被阻塞条件:前驱节点的waitStatus为-1),防止无限循环浪费资源。具体两个方法下面细细分析 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } } // java.util.concurrent.locks.AbstractQueuedSynchronizer // setHead方法是把当前节点置为虚节点,但并没有修改waitStatus,因为它是一直需要用的数据。 private void setHead(Node node) { head = node; node.thread = null; node.prev = null; } // java.util.concurrent.locks.AbstractQueuedSynchronizer // 靠前驱节点判断当前线程是否应该被阻塞 private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { // 获取头结点的节点状态 int ws = pred.waitStatus; // 说明头结点处于唤醒状态 if (ws == Node.SIGNAL) return true; // 通过枚举值我们知道waitStatus>0是取消状态 if (ws > 0) { do { // 循环向前查找取消节点,把取消节点从队列中剔除 node.prev = pred = pred.prev; } while (pred.waitStatus > 0); pred.next = node; } else { // 设置前任节点等待状态为SIGNAL compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } return false; } // java.util.concurrent.locks.AbstractQueuedSynchronizer // parkAndCheckInterrupt主要用于挂起当前线程,阻塞调用栈,返回当前线程的中断状态。 private final boolean parkAndCheckInterrupt() { LockSupport.park(this); return Thread.interrupted(); }

上述方法的流程图如下:

从上图可以看出,跳出当前循环的条件是当“前置节点是头结点,且当前线程获取锁成功”。为了防止因死循环导致CPU资源被浪费,我们会判断前置节点的状态来决定是否要将当前线程挂起,具体挂起流程用流程图表示如下(shouldParkAfterFailedAcquire流程):

CANCELLED状态节点生成

对acquireQueued方法代码展开分析

讯享网// java.util.concurrent.locks.AbstractQueuedSynchronizer final boolean acquireQueued(final Node node, int arg) { boolean failed = true; try { ... for (;;) { final Node p = node.predecessor(); if (p == head && tryAcquire(arg)) { ... failed = false; ... } ... } finally { if (failed) cancelAcquire(node); } } // java.util.concurrent.locks.AbstractQueuedSynchronizer // 通过cancelAcquire方法,将Node的状态标记为CANCELLED private void cancelAcquire(Node node) { // 将无效节点过滤 if (node == null) return; // 设置该节点不关联任何线程,也就是虚节点 node.thread = null; Node pred = node.prev; // 通过前驱节点,跳过取消状态的node while (pred.waitStatus > 0) node.prev = pred = pred.prev; // 获取过滤后的前驱节点的后继节点 Node predNext = pred.next; // 把当前node的状态设置为CANCELLED node.waitStatus = Node.CANCELLED; // 如果当前节点是尾节点,将从后往前的第一个非取消状态的节点设置为尾节点 // 更新失败的话,则进入else,如果更新成功,将tail的后继节点设置为null if (node == tail && compareAndSetTail(node, pred)) { compareAndSetNext(pred, predNext, null); } else { int ws; // 如果当前节点不是head的后继节点,1:判断当前节点前驱节点的是否为SIGNAL,2:如果不是,则把前驱节点设置为SINGAL看是否成功 // 如果1和2中有一个为true,再判断当前节点的线程是否为null // 如果上述条件都满足,把当前节点的前驱节点的后继指针指向当前节点的后继节点 if (pred != head && ((ws = pred.waitStatus) == Node.SIGNAL || (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) && pred.thread != null) { Node next = node.next; if (next != null && next.waitStatus <= 0) compareAndSetNext(pred, predNext, next); } else { // 如果当前节点是head的后继节点,或者上述条件不满足,那就唤醒当前节点的后继节点 unparkSuccessor(node); } node.next = node; // help GC } }
  • 获取当前节点的前驱节点,如果前驱节点的状态是CANCELLED,那就一直往前遍历,找到第一个waitStatus <= 0的节点,将找到的Pred节点和当前Node关联,将当前Node设置为CANCELLED。
  • 根据当前节点的位置,考虑以下三种情况:当前节点是尾节点;当前节点是Head的后继节点;当前节点不是Head的后继节点,也不是尾节点。

如何解锁

由于ReentrantLock在解锁的时候,并不区分公平锁和非公平锁,所以我们直接看解锁的源码展开分析:

// java.util.concurrent.locks.ReentrantLock public void unlock() { sync.release(1); } // java.util.concurrent.locks.AbstractQueuedSynchronizer public final boolean release(int arg) { if (tryRelease(arg)) { Node h = head; if (h != null && h.waitStatus != 0) unparkSuccessor(h); return true; } return false; } // java.util.concurrent.locks.ReentrantLock.Sync // 方法返回当前锁是不是没有被线程持有 protected final boolean tryRelease(int releases) { // 减少可重入次数 int c = getState() - releases; // 当前线程不是持有锁的线程,抛出异常 if (Thread.currentThread() != getExclusiveOwnerThread()) throw new IllegalMonitorStateException(); boolean free = false; // 如果持有线程全部释放,将当前独占锁所有线程设置为null,并更新state if (c == 0) { free = true; setExclusiveOwnerThread(null); } setState(c); return free; } // java.util.concurrent.locks.AbstractQueuedSynchronizer public final boolean release(int arg) { // 上边自定义的tryRelease如果返回true,说明该锁没有被任何线程持有 if (tryRelease(arg)) { // 获取头结点 Node h = head; // 头结点不为空并且头结点的waitStatus不是初始化节点情况,解除线程挂起状态 if (h != null && h.waitStatus != 0) unparkSuccessor(h); return true; } return false; } // java.util.concurrent.locks.AbstractQueuedSynchronizer private void unparkSuccessor(Node node) { // 获取头结点waitStatus int ws = node.waitStatus; if (ws < 0) compareAndSetWaitStatus(node, ws, 0); // 获取当前节点的下一个节点 Node s = node.next; // 如果下个节点是null或者下个节点被cancelled,就找到队列最开始的非cancelled的节点 if (s == null || s.waitStatus > 0) { s = null; // 就从尾部节点开始找,到队首,找到队列第一个waitStatus<0的节点。 for (Node t = tail; t != null && t != node; t = t.prev) if (t.waitStatus <= 0) s = t; } // 如果当前节点的下个节点不为空,而且状态<=0,就把当前节点unpark if (s != null) LockSupport.unpark(s.thread); }


综上所述,如果是从前往后找,由于极端情况下入队的非原子操作和CANCELLED节点产生过程中断开Next指针的操作,可能会导致无法遍历所有的节点。所以,唤醒对应的线程后,对应的线程就会继续往下执行。

中断恢复后的执行流程

唤醒后,会执行return Thread.interrupted();这个函数返回的是当前执行线程的中断状态,并清除。

讯享网// java.util.concurrent.locks.AbstractQueuedSynchronizer private final boolean parkAndCheckInterrupt() { LockSupport.park(this); return Thread.interrupted(); }

回到acquireQueued代码,当parkAndCheckInterrupt返回True或者False的时候,interrupted的值不同,但都会执行下次循环。如果这个时候获取锁成功,就会把当前interrupted返回。

// java.util.concurrent.locks.AbstractQueuedSynchronizer final boolean acquireQueued(final Node node, int arg) { boolean failed = true; try { boolean interrupted = false; for (;;) { final Node p = node.predecessor(); if (p == head && tryAcquire(arg)) { setHead(node); p.next = null; // help GC failed = false; return interrupted; } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } }

如果acquireQueued为True,就会执行selfInterrupt方法。

讯享网// java.util.concurrent.locks.AbstractQueuedSynchronizer   static void selfInterrupt() {     Thread.currentThread().interrupt(); }

该方法其实是为了中断线程。

六、独占模式源码讲解

下面我们学习下同步的具体运行机制,为了更好的演示,我们用ReentrantLock作为使用入口,一步步跟进源码探究AQS底层是如何运作的,这里说明一下,因为ReentrantLock底层调用的AQS是独占模式,所以下文讲解的AQS源码也是针对独占模式的操作

6.1 加锁过程
final void lock() { if (compareAndSetState(0, 1)) //设置持有锁线程 setExclusiveOwnerThread(Thread.currentThread()); else acquire(1); } 

逻辑很简单,线程进来后直接利用CAS尝试抢占锁,如果抢占成功state值回被改为1,且设置对象独占锁线程为当前线程,否则就调用acquire(1)再次尝试获取锁。

我们假定有两个线程A和B同时竞争锁,A进来先抢占到锁,此时的AQS模型图就类似这样:

继续走下面的方法

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

acquire包含了几个函数的调用,

  • tryAcquire:尝试直接获取锁,如果成功就直接返回;
  • addWaiter:将该线程加入等待队列FIFO的尾部,并标记为独占模式;
  • acquireQueued:线程阻塞在等待队列中获取锁,一直获取到资源后才返回。如果在整个等待过程中被中断过,则返回true,否则返回false。
  • selfInterrupt:自我中断,就是既拿不到锁,又在等待时被中断了,线程就会进行自我中断selfInterrupt(),将中断补上。

我们一个个来看源码,并结合上面的两个线程来做场景分析。

6.1.1 tryAcquire

不用多说,就是为了再次尝试获取锁

protected final boolean tryAcquire(int acquires) { return nonfairTryAcquire(acquires); } final boolean nonfairTryAcquire(int acquires) { final Thread current = Thread.currentThread(); int c = getState(); if (c == 0) { if (compareAndSetState(0, acquires)) { setExclusiveOwnerThread(current); 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; } 

当线程B进来后,nonfairTryAcquire方法首先会获取state的值,如果为0,则正常获取该锁,不为0的话判断是否是当前线程占用了,是的话就累加state的值,这里的累加也是为了配合释放锁时候的次数,从而实现可重入锁的效果。

当然,因为之前锁已经被线程A占领了,所以这时候tryAcquire会返回false,继续下面的流程。

6.1.2 addWaiter :抢锁失败,CLH入队
讯享网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(node); return node; } 

这段代码首先会创建一个和当前线程绑定的Node节点,Node为双向链表。此时等待队列中的tail指针为空,直接调用enq(node)方法将当前线程加入等待队列尾部,然后返回当前结点的前驱结点。

private Node enq(final Node node) { // CAS"自旋",直到成功加入队尾 for (;;) { Node t = tail; if (t == null) { // 队列为空,初始化一个Node结点作为Head结点,并将tail结点也指向它 if (compareAndSetHead(new Node())) tail = head; } else { // 把当前结点插入队列尾部 node.prev = t; if (compareAndSetTail(t, node)) { t.next = node; return t; } } } } 

第一遍循环时,tail指针为空,初始化一个Node结点,并把head和tail结点都指向它,然后第二次循环进来之后,tail结点不为空了,就将当前的结点加入到tail结点后面,也就是这样:

如果此时有另一个线程C进来的话,发现锁已经被A拿走了,然后队列里已经有了线程B,那么线程C就只能乖乖排到线程B的后面去。

 6.1.3 acquireQueued

一旦加入同步队列,就需要使用该方法,自旋阻塞 唤醒来不断的尝试获取锁,直到被中断或获取到锁。

接着解读方法,通过tryAcquire()和addWaiter(),我们的线程还是没有拿到资源,并且还被排到了队列的尾部,如果让你来设计的话,这个时候你会怎么处理线程呢?其实答案也很简单,能做的事无非两个:

1、循环让线程再抢资源。但仔细一推敲就知道不合理,因为如果有多个线程都参与的话,你抢我也抢只会降低系统性能

2、进入等待状态休息,直到其他线程彻底释放资源后唤醒自己,自己再拿到资源

讯享网final boolean acquireQueued(final Node node, int arg) { boolean failed = true; try { // 标记是否会被中断 boolean interrupted = false; // CAS自旋 for (;;) { // 获取当前结点的前结点 final Node p = node.predecessor(); if (p == head && tryAcquire(arg)) { setHead(node); p.next = null; // help GC failed = false; return interrupted; } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) // 获取锁失败,则将此线程对应的node的waitStatus改为CANCEL cancelAcquire(node); } } private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { int ws = pred.waitStatus; if (ws == Node.SIGNAL) // 前驱结点等待状态为"SIGNAL",那么自己就可以安心等待被唤醒了 return true; if (ws > 0) { /* * 前驱结点被取消了,通过循环一直往前找,直到找到等待状态有效的结点(等待状态值小于等于0) , * 然后排在他们的后边,至于那些被当前Node强制"靠后"的结点,因为已经被取消了,也没有引用链, * 就等着被GC了 */ do { node.prev = pred = pred.prev; } while (pred.waitStatus > 0); pred.next = node; } else { // 如果前驱正常,那就把前驱的状态设置成SIGNAL compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } return false; } private final boolean parkAndCheckInterrupt() { LockSupport.park(this); return Thread.interrupted(); } 

acquireQueued方法的流程是这样的:

1、CAS自旋,先判断当前传入的Node的前结点是否为head结点,是的话就尝试获取锁,获取锁成功的话就把当前结点置为head,之前的head置为null(方便GC),然后返回

2、如果前驱结点不是head或者加锁失败的话,就调用shouldParkAfterFailedAcquire,将前驱节点的waitStatus变为了SIGNAL=-1,最后执行parkAndChecknIterrupt方法,调用LockSupport.park()挂起当前线程,parkAndCheckInterrupt在挂起线程后会判断线程是否被中断,如果被中断的话,就会重新跑acquireQueued方法的CAS自旋操作,直到获取资源。

ps:LockSupport.park方法会让当前线程进入waitting状态,在这种状态下,线程被唤醒的情况有两种,一是被unpark(),二是被interrupt(),所以,如果是第二种情况的话,需要返回被中断的标志,然后在acquire顶层方法的窗口那里自我中断补上

此时,因为线程A还未释放锁,所以线程B状态都是被挂起的

到这里,加锁的流程就分析完了,其实整体来说也并不复杂,而且当你理解了独占模式加锁的过程,后面释放锁和共享模式的运行机制也没什么难懂的了,所以整个加锁的过程还是有必要多消化下的,也是AQS的重中之重。

为了方便你们更加清晰理解,我加多一张流程图吧

6.2 释放锁

说完了加锁,我们来看看释放锁是怎么做的,AQS中释放锁的方法是release(),当调用该方法时会释放指定量的资源 (也就是锁) ,如果彻底释放了(即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; } 
6.2.1 tryRelease

代码上可以看出,核心的逻辑都在tryRelease方法中,该方法的作用是释放资源,AQS里该方法没有具体的实现,需要由自定义的同步器去实现,我们看下ReentrantLock代码中对应方法的源码:

讯享网protected final boolean tryRelease(int releases) { 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; } 

tryRelease方法会减去state对应的值,如果state为0,也就是已经彻底释放资源,就返回true,并且把独占的线程置为null,否则返回false。

此时AQS中的数据就会变成这样:

完全释放资源后,当前线程要做的就是唤醒CLH队列中第一个在等待资源的线程,也就是head结点后面的线程,此时调用的方法是unparkSuccessor()

private void unparkSuccessor(Node node) { int ws = node.waitStatus; if (ws < 0) //将head结点的状态置为0 compareAndSetWaitStatus(node, ws, 0); //找到下一个需要唤醒的结点s Node s = node.next; //如果为空或已取消 if (s == null || s.waitStatus > 0) { s = null; // 从后向前,直到找到等待状态小于0的结点,前面说了,结点waitStatus小于0时才有效 for (Node t = tail; t != null && t != node; t = t.prev) if (t.waitStatus <= 0) s = t; } // 找到有效的结点,直接唤醒 if (s != null) LockSupport.unpark(s.thread);//唤醒 } 
讯享网for (;;) { // 获取当前结点的前结点 final Node p = node.predecessor(); if (p == head && tryAcquire(arg)) { setHead(node); p.next = null; // help GC failed = false; return interrupted; } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } 

当线程B获取锁之后,会把当前结点赋值给head,然后原先的前驱结点 (也就是原来的head结点) 去掉引用链,方便回收,这样一来,线程B获取锁的整个过程就完成了,此时AQS的数据就会变成这样。

 到这里,我们已经分析完了AQS独占模式下加锁和释放锁的过程,也就是tryAccquire->tryRelease这一链条的逻辑,除此之外,AQS中还支持共享模式的同步,这种模式下关于锁的操作核心其实就是tryAcquireShared->tryReleaseShared这两个方法,我们可以简单看下。

七. 共享模式 源码讲解

7.1 获取锁

AQS中,共享模式获取锁的顶层入口方法是acquireShared,该方法会获取指定数量的资源,成功的话就直接返回,失败的话就进入等待队列,直到获取资源。

public final void acquireShared(int arg) { if (tryAcquireShared(arg) < 0) doAcquireShared(arg); } 

该方法里包含了两个方法的调用,

tryAcquireShared:尝试获取一定资源的锁,返回的值代表获取锁的状态。

doAcquireShared:进入等待队列,并循环尝试获取锁,直到成功。

7.1.1 tryAcquireShared

tryAcquireShared在AQS里没有实现,同样由自定义的同步器去完成具体的逻辑,像一些较为常见的并发工具Semaphore、CountDownLatch里就有对该方法的自定义实现,虽然实现的逻辑不同,但方法的作用是一样的,就是获取一定资源的资源,然后根据返回值判断是否还有剩余资源,从而决定下一步的操作。

返回值有三种定义:

  • 负值代表获取失败;
  • 0代表获取成功,但没有剩余的资源,也就是state已经为0;
  • 正值代表获取成功,而且state还有剩余,其他线程可以继续领取。

当返回值小于0时,证明此次获取一定数量的锁失败了,然后就会走doAcquireShared方法

7.1.2 doAcquireShared

此方法的作用是将当前线程加入等待队列尾部休息,直到其他线程释放资源唤醒自己,自己成功拿到相应量的资源后才返回,这是它的源码:

讯享网private void doAcquireShared(int arg) { // 加入队列尾部 final Node node = addWaiter(Node.SHARED); boolean failed = true; try { boolean interrupted = false; // CAS自旋 for (;;) { final Node p = node.predecessor(); // 判断前驱结点是否是head if (p == head) { // 尝试获取一定数量的锁 int r = tryAcquireShared(arg); if (r >= 0) { // 获取锁成功,而且还有剩余资源,就设置当前结点为head,并继续唤醒下一个线程 setHeadAndPropagate(node, r); // 让前驱结点去掉引用链,方便被GC p.next = null; // help GC if (interrupted) selfInterrupt(); failed = false; return; } } // 跟独占模式一样,改前驱结点waitStatus为-1,并且当前线程挂起,等待被唤醒 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } } private void setHeadAndPropagate(Node node, int propagate) { Node h = head; // head指向自己 setHead(node); // 如果还有剩余量,继续唤醒下一个邻居线程 if (propagate > 0 || h == null || h.waitStatus < 0) { Node s = node.next; if (s == null || s.isShared()) doReleaseShared(); } } 

看到这里,你会不会一点熟悉的感觉,这个方法的逻辑怎么跟上面那个acquireQueued() 那么类似啊?对的,其实两个流程并没有太大的差别。只是doAcquireShared()比起独占模式下的获取锁上多了一步唤醒后继线程的操作,当获取完一定的资源后,发现还有剩余的资源,就继续唤醒下一个邻居线程,这才符合"共享"的思想嘛。

这里我们可以提出一个疑问,共享模式下,当前线程释放了一定数量的资源,但这部分资源满足不了下一个等待结点的需要的话,那么会怎么样?

按照正常的思维,共享模式是可以多个线程同时执行的才对,所以,多个线程的情况下,如果老大释放完资源,但这部分资源满足不了老二,但能满足老三,那么老三就可以拿到资源。可事实是,从源码设计中可以看出,如果真的发生了这种情况,老三是拿不到资源的,因为等待队列是按顺序排列的,老二的资源需求量大,会把后面量小的老三以及老四、老五等都给卡住。从这一个角度来看,虽然AQS严格保证了顺序,但也降低了并发能力

接着往下说吧,唤醒下一个邻居线程的逻辑在doReleaseShared()中,我们放到下面的释放锁来解析。

7.2 释放锁

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

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

该方法同样包含两部分的逻辑:

tryReleaseShared:释放资源。

doReleaseShared:唤醒后继结点。

跟tryAcquireShared方法一样,tryReleaseShared在AQS中没有具体的实现,由子同步器自己去定义,但功能都一样,就是释放一定数量的资源。

释放完资源后,线程不会马上就收工,而是唤醒等待队列里最前排的等待结点。

7.2.1 doReleaseShared

唤醒后继结点的工作在doReleaseShared()方法中完成,我们可以看下它的源码:

讯享网private void doReleaseShared() { for (;;) { // 获取等待队列中的head结点 Node h = head; if (h != null && h != tail) { int ws = h.waitStatus; // head结点waitStatus = -1,唤醒下一个结点对应的线程 if (ws == Node.SIGNAL) { if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) continue; // loop to recheck cases // 唤醒后继结点 unparkSuccessor(h); } else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) continue; // loop on failed CAS } if (h == head) // loop if head changed break; } } 

代码没什么特别的,就是如果等待队列head结点的waitStatus为-1的话,就直接唤醒后继结点,唤醒的方法unparkSuccessor()在上面已经讲过了,这里也没必要再复述。

八、AQS原理总结

AQS的原理可以简单概括为以下几点:

  1. AQS维护了一个FIFO队列,用于存储处于等待状态的线程。当一个线程需要获取某个资源时,如果该资源已被其他线程占用,则该线程会被加入到等待队列中,等待其他线程释放该资源。
  2. AQS使用一个state字段来表示资源的状态,state字段可以被多个线程同时访问。当一个线程需要获取某个资源时,它会先尝试修改state字段的值,以此来表示它已经获取了该资源。如果该资源已经被其他线程占用,则该线程会被加入到等待队列中,等待其他线程释放该资源。
  3. AQS中提供了两种模式:独占模式和共享模式。独占模式表示只有一个线程可以同时访问某个资源,比如ReentrantLock就是一个独占模式的同步器。共享模式则表示多个线程可以同时访问某个资源,比如Semaphore就是一个共享模式的同步器。
  4. AQS中还提供了一些钩子方法,这些方法可以被子类重写以实现自定义同步器的行为,比如acquire、release、tryAcquire等方法。

AQS的实现原理其实就是上述几个核心点的实现。对于AQS的使用者来说,只需要了解AQS的使用方法即可。而对于AQS的实现者来说,需要深入理解AQS的实现原理,并且了解AQS在不同场景下的具体应用,以便于在需要的时候能够根据具体需求来实现自定义的同步器。

总之,AQS是Java并发编程中的一个重要组件,可以帮助我们更方便地实现线程间的协调和互斥,同时也是Java并发编程中的一个基础知识点,需要深入理解和掌握。


大厂面试问题


【1】什么是 AQS? 为什么它是核心?
【2】AQS的核心思想是什么? 它是怎么实现的? 底层数据结构等?
【3】AQS有哪些核心的方法?
【4】AQS定义什么样的资源获取方式? AQS定义了两种资源获取方式:独占(只有一个线程能访问执行,又根据是否按队列的顺序分为公平锁和非公平锁,如ReentrantLock) 和共享(多个线程可同时访问执行,如Semaphore、CountDownLatch、 CyclicBarrier )。ReentrantReadWriteLock可以看成是组合式,允许多个线程同时对某一资源进行读。
【5】AQS底层使用了什么样的设计模式?
 

小讯
上一篇 2025-04-02 15:09
下一篇 2025-01-06 09:49

相关推荐

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容,请联系我们,一经查实,本站将立刻删除。
如需转载请保留出处:https://51itzy.com/kjqy/59326.html