AQS源码分析

原创

BNTang

发布于 2023-09-30 19:51:04

2050

发布于 2023-09-30 19:51:04

文章被收录于专栏：『学习与分享之旅』『学习与分享之旅』

AQS 概述

什么是 AQS

AQS 是 java.util.concurrent.locks.AbstractQueuedSynchronizer 类的缩写，抽象的队列同步器 AQS 定义了一套多线程访问共享资源的同步器框架，许多同步类实现都依赖于它 AQS 支持独占锁（exclusive）和共享锁（share）两种模式独占锁：只能被一个线程获取到（Reentrantlock）共享锁：可以被多个线程同时获取（CountDownLatch、ReadWriteLock）它维护了一个 volatile int state（代表共享资源）和一个 FIFO 线程等待队列无论是独占锁还是共享锁，本质上都是对 AQS 内部的一个变量 state 的获取。state 是一个原子的 int 变量，用来表示锁状态、资源数等 JDK 中许多并发工具类的内部实现都依赖于 AQS，如 ReentrantLock，Semaphore，CountDownLatch 等等

AQS 作者

Doug Lea 道格.李

ReentrantLock 源码 Lock 分析

类概述

ReentrantLock 是一种基于 AQS 框架的应用实现，是 JDK 中的一种线程并发访问的同步手段，它的功能类似于 synchronized 是一种互斥锁，可以保证线程安全

它具有比 synchronized 更多的特性，比如它支持手动加锁与解锁，支持加锁的公平性

类内部的继承结构

ReentrantLock 通过定义了一个内部类 Sync 去继承 AbstractQueuedSynchronizer

进入到 AbstractQueuedSynchronizer 选中类名 ctrl + shift + alt + u 查看继承结构图

查看实现类

ctrl + A 全选回车

类图分析

在 ReentrantLock 内部定义了一个 Sync 的内部类，该类继承 AbstractQueuedSynchronized，对该抽象类的部分方法做了实现，并且还定义了两个子类

FairSync：公平锁的实现
NonfairSync：非公平锁的实现

Lock 源码方法

基于 公平锁 进行查看

非公平锁一上来就尝试获取到锁，获取不到就进入到队列当中，在调用 tryAcquire 时采用了 模板方法设计模式，调用 tryAcquire 时调用的是实现类 FairSync 当中的 tryAcquire 当中的方法，如下图

自己类当中定义的方法仅仅是做一个占位，里面是直接抛出了一个异常，AQS 只是一个框架，具体资源的获取 / 释放方式交由自定义同步器去实现

开始尝试获得锁

此方法为核心方法，通过此方法来获取锁，下面会详细讲解内部实现原理

如上图获取锁失败后，会进入到队列当中

方法调用泳道图

CLH 队列

CLH 同步队列

叫做 CLH(Craig，Landin，and Hagersten) 当线程要入队的时候，只需要构建成一个 Node 节点加入队尾即可 CLH 队列可以看成是一个双向链表实现的，刚开始队列里面只有一个空的 Head 节点，这个时候 Head 节点和 Tail 节点都是同一个节点如果要出队，只需要将队首节点移除，同时唤醒下一个等待的节点同时更新节点之间的引用关系然后加入到队列里面后，一直自旋去尝试获取同步等待状态

AQS 通过内置的 FIFO 同步双向队列来完成资源获取线程的排队工作，内部通过节点 head【实际上是虚拟节点，真正的第一个线程在 head.next 的位置】和 tail 记录队首和队尾元素，队列元素类型为 Node。

Node

// 表示一个共享节点模式下等待的常量
static final Node SHARED = new Node();

// 表示在互斥模式下等待的常量
static final Node EXCLUSIVE = null;

// 表示当前节点线程被取消
static final int CANCELLED = 1;

// 表示当前节点线程等待被唤醒
static final int SIGNAL = -1;

// 表示当前节点线程正在等待某个条件
static final int CONDITION = -2;

// 表示下一次共享模式的同步等待状态的获取将会一直循环下去
static final int PROPAGATE = -3;

// 等待状态
volatile int waitStatus;

// 当前节点的前一个节点
volatile Node prev;

// 当前节点的后一个节点
volatile Node next;

// 获取同步等待状态的当前线程
volatile Thread thread;

// 当前节点的下一个处于等待状态的节点
Node nextWaiter;

过程

如果当前线程获取同步状态失败（锁）时，AQS 则会将当前线程以及等待状态等信息构造成一个节点（Node）并将其加入同步队列，同时会阻塞当前线程，当同步状态释放时，则会把节点中的线程唤醒，使其再次尝试获取同步状态，AQS 主要就靠这个 Node 节点来维护同步等待队列。

AQS 实现原理

它维护了一个 volatile int state（代表共享资源，用来表示锁状态、资源数）和一个 FIFO 线程等待队列

state 默认初始化为 0，当为 0 的时候，代表当中这个锁资源没有被人持有，可以进行加锁，每加 1，代表重入一次

阻塞线程

在 AQS 的实现中有一个出现了一个 park 的概念
park 即 LockSupport.park() 它的作用是阻塞当前线程，并且可以调用 LockSupport.unpark(Thread) 去停止阻塞
它们的实质都是通过 UnSafe 类使用了 CPU 的原语
在 AQS 中使用 park 的主要作用是，让排队的线程阻塞掉（停止其自旋，自旋会消耗 CPU 资源）并在需要的时候，可以方便的唤醒阻塞掉的线程

AQS 源码

在创建 ReentrantLock 将参数去掉，也就是不传入参数内部创建的锁为非公平锁。

直接按住键盘上的 Ctrl + 鼠标左键进入 ReentrantLock 构造器查看就可以发现了如下图。

如果在创建时添加了参数。

如果是一个非公平锁，会上来就先去抢占锁资源，公平锁的话，会先进行排队。

如果等于 0，说明没有人加锁，直接使用 CAS 进行加锁加锁之后，设置当前线程拥有这把锁，为独占的线程如果当前线程已经占有了该锁，就对 state 加 1，说明是重入的 (重入锁)

如上截图的方法含义分别如下

获取当前线程
取出 state 同步器状态值，看看是否为 0
- 如果为 0 代表当前同步器允许被线程占有
- 当前是一个无锁的状态，可以进行加锁
如果为 0，可能后面还有人再排队等待
- 由于使用的是公平锁
- 所以要先判断下有没有人在排队等待
- 判断方式请参考下图中的方法里面是具体的实现判断代码 (JDK8)

如果没有人排队，就通过 CAS 的方式加锁
把 state 变成 1
并且 exclusiveOwnThread 引用着当前线程

如果不等于 0
- 判断当前持有同步器的锁是不是当前的线程，exclusiveOwnThread 引用是否为当前线程
- 如果是当前线程的话，就把 state 再加 1，同一个线程做重入
- 如果不是自己，就返回 false 代表加锁失败

抢占资源失败后，进入到队列中进行等待。

在构造器当中记录当前线程，把当前线程的模式传进来，记录当前线程是互斥的，还是共享的，记录当前的线程

创建节点后，开启一个自旋，保证入队成功
第一次，尾部节点为空，进入到初始化队列中

使用 VarHandler 的方式来进行 CAS 操作如果直接拿头节点属性的话，没有办法直接使用 CAS 可以完成线程安全的操作从 JDK9 之后增加的内容是第一个节点，初始化头部队列与尾部队列，两个是同一个由于是多线程同时进行加锁，所以要考虑线程安全如果直接锁定整个链表的话，锁的粒度就比较大，这里面加锁采用的是 CAS 加锁的方式使用 CAS 保证入队安全，获取原来的 tail，在修改前进对比，看看是不是原来那个了，如果不是重新获取，获取之后再修改如果尾节点不为空，尾部节点设置为当前节点的上一个节点

把原来尾部的节点的下一个节点，设置为当前节点，成功入队，并返回当前入队的节点入队之后，线程进入阻塞状态，尝试获取锁

如果前置节点不是头节点或者前置节点竞争失败，就进入阻塞。

获取当前线程的 waitStatus 的值，如果为 -1，就阻塞如果不是 -1，就设置为 -1，下一次再调用的时候，返回 true，阻塞线程

waitStatus

等待状态值	概述
int CANCELLED = 1;	由于超时或中断，此节点被取消。节点一旦被取消了就不会再改变状态。特别是，取消节点的线程不会再阻塞
int SIGNAL = -1;	当前结点为 `-1`，则说明后一个结点需要 `park` 阻塞，只有上一个节点是 `SIGNAL`，当前节点才有可能被上一个节点唤醒
int CONDITION = -2;	条件队列
int PROPAGATE = -3;	传播

释放锁

把状态减 1 清空对当前 state 的独占

中断等待

使用 lock 不会中断等待。

入队时，会清空中断状态 Thread.interupted

使用 lockInterruptibly 会中断等待，可中断线程源码如下。

获取到线程中断后，直接抛出一个异常，取消线程的竞争，设置状态为 NODE.CANCELLED 状态，把节点从队列当中移除，重新建立前后节点，获取前驱节点 cancelAcquire(node)

private void cancelAcquire(Node node) {
    // 如果为空,返回
    if (node == null)
        return;
    // 清空node中的线程
    node.thread = null;
    // 获取当前节点的前一个节点
    Node pred = node.prev;
    // 通过循环找出,找一个可用的节点,>0的都是为取消的
    while (pred.waitStatus > 0)
        node.prev = pred = pred.prev;
    // 记录后一个节点指向的后一个节点
    Node predNext = pred.next;
    // 把当前节点设置为Node.CANCELLED,设置为1
    node.waitStatus = Node.CANCELLED;
    // 如果当前节点是尾节点,就把前一个节点,设置为尾节点为
    if (node == tail && compareAndSetTail(node, pred)) {
        // 并且所尾部节点的next设置为null
        pred.compareAndSetNext(predNext, null);
    } else {
        // 不是尾节点,或者是尾节点,但是CAS失败
        int ws;
        // 前一个节点不头节点,记录前一个节点的ws, 并判断是否为-1
        // 如果不是-1 就把它的设置为-1,并且线程不为空
        if (pred != head
            && ((ws = pred.waitStatus) == Node.SIGNAL || (ws <= 0 && pred.compareAndSetWaitStatus(ws, Node.SIGNAL)))
            && pred.thread != null) {
            // 记录当前节点的节一个节点
            Node next = node.next;
            // 下一个节点不为空,并且下一个节点ws<=0
            if (next != null && next.waitStatus <= 0)
                // 把当前节点的下一个节点,与当前节点的上一个节点,建立关系
                pred.compareAndSetNext(predNext, next);
        } else {
            // 唤醒当前节点
            unparkSuccessor(node);
        }
        // 把当前节点下一个节点指向它自己
        node.next = node; // help GC
    }
}

Semaphore 源码

acquire 方法

默认是可中断的

获取锁

如果小于 1，排队等待如果大于 1，获取 state 的值拿可用的 state 值 - 当前获取的值如果得到的结果 < 0，则等待不小于 0，则把可用的值，设置为减的结果

重新获取锁，如果大于 0，说明获取得了锁执行 setHeadAndPropagate

AQS原理图：https://www.cnblogs.com/BNTang/gallery/image/385879.html

我正在参与2023腾讯技术创作特训营第二期有奖征文，瓜分万元奖池和键盘手表

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

2023腾讯·技术创作特训营第二期

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

2023腾讯·技术创作特训营第二期

登录后参与评论

0 条评论

热度