AQS源码分析

news/2024/4/27 3:04:44

前言

AbstractQueuedSynchronizer是抽象同步队列,其是实现同步机器的基础组件,并发包中的锁的底层就是使用AQS实现的。

AQS中 维护了一个volatile int state(代表共享资源)和一个FIFO线程等待队列(多线程争用资源被阻塞时会进入此队列)。
这里volatile能够保证多线程下的可见性,当state=1则代表当前对象锁已经被占有,其他线程来加锁时则会失败,加锁失败的线程会被放入一个FIFO的等待队列中,并且会被UNSAFE.park()操作挂起,等待其他获取锁的线程释放锁才能够被唤醒。
另外state的操作都是通过CAS来保证其并发修改的安全性。

一、AQS中的关键成员变量

  • state
    • 在AQS中,维护了一个单一变量state,对于不同的实现其有不同的意义:
    • 在ReentrantLock中,state表示重入式锁的可重入次数
    • 在ReentrantReadWriteLock中,state的高16位用于表示读锁的可获取次数,低16位用于表示写锁的可重入次数。
  • exclusiveOwnerThread
    • 继承自AbstractOwnableSynchronizer,用于指明当前独占线程。
  • headtail
    • 维护了一个队列,分别指向首尾节点
  • Node
    • Node节点内部的SHARED用来标记该线程是在获取共享资源时被阻塞挂起放入AQS队列的,EXCLUSIVE用来标识该线程是获取独占资源时被阻塞挂起放入AQS队列的。
    • 在Node节点内部有一个成员变量waitStatus记录当前线程等待状态,可以为:
      • 1:CANCELLED(线程被取消了)
      • -1:SIGNAL(线程需要唤醒)
      • -2:CONDITION(线程在条件队列里等待)
      • -3:PROPAGATE(释放资源时需要通知其他节点)
  • ConditionObject
    • ConditionObject和Node一样是AQS的内部类。它用来结合锁实现线程同步,其可以访问AQS的内部变量(state和AQS阻塞队列)。
    • ConditionObject是条件变量,每个条件变量对应一个条件队列,我们可以看到ConditionObject中有两个指针,分别指向条件队列的队尾和队头。条件队列用来存放调用条件变量的await方法后被阻塞的线程。

二、线程中断相关的三个方法

三、Unsafe与LockSupport

Unsafe

  • CAS的全称是Compare-And-Swap,它是一条CPU并发原语。
  • 它的功能是判断内存某个位置是否是预期值,如果是则更改为新的值,这个过程是原子性的
  • CAS并发原语在 java的体现就是sun.mic.Unsafe类个各个方法,调用Unsafe类的方法,JVM会帮助我们实现CAS汇编指令。这是一个完全依赖于硬件的功能,通过它实现原子性操作。由于CAS是一种系统原语,由若干指令组成,该原语执行必须连续的不许中断。

这里设置了静态代码块提前获取了state、head、tail、waitStatus、next四个参数在对象内存中的偏移量。

    private static final Unsafe unsafe = Unsafe.getUnsafe();private static final long stateOffset;private static final long headOffset;private static final long tailOffset;private static final long waitStatusOffset;private static final long nextOffset;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); }}/*** CAS head field. Used only by enq.*/private final boolean compareAndSetHead(Node update) {return unsafe.compareAndSwapObject(this, headOffset, null, update);}/*** CAS tail field. Used only by enq.*/private final boolean compareAndSetTail(Node expect, Node update) {return unsafe.compareAndSwapObject(this, tailOffset, expect, update);}/*** CAS waitStatus field of a node.*/private static final boolean compareAndSetWaitStatus(Node node,int expect,int update) {return unsafe.compareAndSwapInt(node, waitStatusOffset,expect, update);}/*** CAS next field of a node.*/private static final boolean compareAndSetNext(Node node,Node expect,Node update) {return unsafe.compareAndSwapObject(node, nextOffset, expect, update);}

LockSupport

LockSupport是用来创建锁和其他同步类的基本线程阻塞原语。简而言之,当调用LockSupport.park时,表示当前线程将会等待,直至获得许可,当调用LockSupport.unpark时,必须把等待获得许可的线程作为参数进行传递,好让此线程继续运行

  • park函数,阻塞线程,并且该线程在下列情况发生之前都会被阻塞: ① 调用unpark函数,释放该线程的许可。② 该线程被中断。③ 设置的时间到了。并且,当time为绝对时间时,isAbsolute为true,否则,isAbsolute为false。当time为0时,表示无限等待,直到unpark发生。
  • unpark函数,释放线程的许可,即激活调用park后阻塞的线程。这个函数不是安全的,调用这个函数时要确保线程依旧存活。
public class LockSupportDemo {public static void main(String[] args) {Thread A = new Thread(new Runnable() {@Overridepublic void run() {System.out.println("线程A被LockSupport.park()阻塞");LockSupport.park();System.out.println("线程A被线程B LockSupport.unpark()唤醒");}},"A");A.start();Thread B = new Thread(new Runnable() {@Overridepublic void run() {System.out.println("线程B唤醒线程A");// 唤醒指定线程t,也就是ALockSupport.unpark(A);}},"B")B.start();}
}结果:
线程ALockSupport.park()阻塞
线程B唤醒线程A
线程A被线程B LockSupport.unpark()唤醒

四、核心源码

以ReentrantLock为例进行讲解,AQS是典型的模板方法的实现,所以AQS对外暴露了多个个抽象方法(tryAcquire、tryRelease等等)需要子类进行实现。
ReentrantLock的lock方法实际上调用了sync的lock方法,而sync继承了AQS,同时针对公平策略和非公平策略有不同的实现。这里我们主要看针对非公平锁NonfairSync的实现。

lock()

    static final class NonfairSync extends Sync {private static final long serialVersionUID = 7316153563782823691L;/*** Performs lock.  Try immediate barge, backing up to normal* acquire on failure.*/final void lock() {if (compareAndSetState(0, 1))setExclusiveOwnerThread(Thread.currentThread());elseacquire(1);}protected final boolean tryAcquire(int acquires) {return nonfairTryAcquire(acquires);}}
    public final void acquire(int arg) {if (!tryAcquire(arg) &&acquireQueued(addWaiter(Node.EXCLUSIVE), arg))selfInterrupt();}
        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) // overflowthrow new Error("Maximum lock count exceeded");setState(nextc);return true;}return false;}
    private Node addWaiter(Node mode) {Node node = new Node(Thread.currentThread(), mode);// Try the fast path of enq; backup to full enq on failureNode pred = tail;// 获取锁失败,再次判断队列是否初始化if (pred != null) {node.prev = pred;if (compareAndSetTail(pred, node)) {pred.next = node;return node;}}enq(node);return node;}
    private Node enq(final Node node) {// 第一次执行,也就是head和tail两个指针都为null,会初始化两个Nodefor (;;) {Node t = tail;// 初始化队列,设置一个空Node,并将head与tail两个指针同时指向该节点if (t == null) { // Must initializeif (compareAndSetHead(new Node()))tail = head;// 队列已经初始化完成,则将该节点插入队列尾部} else {node.prev = t;// 注意,此时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();// 如果是第一次入队,则再次尝试获取stateif (p == head && tryAcquire(arg)) {setHead(node);p.next = null; // help GCfailed = false;return interrupted;}// 非第一次入队/第一次入对的第二次循环if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())interrupted = true;}} finally {if (failed)cancelAcquire(node);}}

因为lock方法是不可中断的,所以从lock方法中进来构建起来的同步队列不会有CANCELLED状态。CONDITION用于条件队列当中。PROPAGETE是用于共享模式下的状态。

    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {// Acquire失败以后是否需要挂起,true:需要-false:不需要// 针对ReentrantLock,这里指挥判断SIGNALint 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;// ws > 0  =  CANCELLEDif (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();}


整体流程:
image.png

unLock()

    public void unlock() {sync.release(1);}
    public final boolean release(int arg) {if (tryRelease(arg)) {Node h = head;if (h != null && h.waitStatus != 0)unparkSuccessor(h);return true;}return false;}

image.png

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

这里线程二被唤醒以后将继续执行acquireQueued方法,判断线程二的前置节点是否为head,如果是则继续使用tryAcquire()方法来尝试获取锁,其实就是使用CAS操作来修改state值,如果修改成功则代表获取锁成功。接着将线程二设置为head节点,然后空置之前的head节点数据,被空置的节点数据等着被垃圾回收

在线程二释放锁以后,这个时候CLH队列中就只剩下线程三:

五、知识拓展

公平锁与非公平锁

非公平锁执行原理:

公平锁执行原理:

参考内容:

  • https://www.bilibili.com/video/BV1vM411r7Bt
  • https://juejin.cn/post/6844904146127044622

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.cpky.cn/p/11146.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈,一经查实,立即删除!

相关文章

01、Lua 入门教程

Lua 入门教程 Lua是一种轻量小巧的脚本语言&#xff0c;用标准C语言编写并以源代码形式开放&#xff0c; 其设计目的是为了嵌入应用程序中&#xff0c;从而为应用程序提供灵活的扩展和定制功能。 Lua是巴西里约热内卢天主教大学&#xff08;Pontifical Catholic University of…

数据结构-----栈、顺序栈、链栈

在软件应用中&#xff0c;栈这种后进先出数据结构的应用是非常普遍的。比如用浏览器上网时&#xff0c;不管什么浏览器都有一个“后退”键&#xff0c;你点击后可以按访问顺序的逆序加载浏览过的网页。即使从一个网页开始&#xff0c;连续点了几十个链接跳转&#xff0c;你点“…

零基础自学C语言|文件操作

✈为什么使用文件&#xff1f; 如果没有文件&#xff0c;我们写的程序的数据是存储在电脑的内存中&#xff0c;如果程序退出&#xff0c;内存回收&#xff0c;数据就丢失了&#xff0c;等再次运行程序&#xff0c;是看不到上次程序的数据的&#xff0c;如果要将数据进行持久化…

python数据分析和可视化【3】体检数据分析和小费数据分析

文章目录 体检数据分析小费数据分析 体检数据分析 要求&#xff1a; &#xff08;1&#xff09;读取testdata文件&#xff0c;利用agg函数统计数据中‘淋巴细胞计数’的和与均值、‘白细胞计数’的和与均值。 &#xff08;2&#xff09;统计不同性别人群的血小板计数 &#xf…

00000基础搭建vue+flask前后端分离项目

我完全是参考的这个vue3flask前后端分离环境速建_flask vue3-CSDN博客 安装了node_js&#xff08;添加了环境变量&#xff09; 环境变量 把原来的镜像源换成了淘宝镜像源 npm config set registry https://registry.npmmirror.com/ 查看版本证明安装成功 npm - v 安装npm i…

client-go中ListAndWatch机制,informer源码详解

文章首发地址&#xff1a; 学一下 (suxueit.com)https://suxueit.com/article_detail/s9UMb44BWZdDRfKqFv22 先上一张&#xff0c;不知道是那个大佬画的图 简单描述一下流程 client-go封装部分 以pod为例 、先List所有的Pod资源&#xff0c;然后通过已经获取的pod资源的最大版…