AQS 抽象队列同步器

2024-01-07 17:52
文章标签 队列 抽象 aqs 同步器

本文主要是介绍AQS 抽象队列同步器，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

AQS

AQS （抽象队列同步器）： AbstractQueuedSynchronizer 是什么

  • 来自jdk1.5，是用来实现锁或者其他同步器组件的公共基础部分的抽象实现，是重量级基础框架以及JUC的基石，主要用于解决锁分配给谁的问题
  • 整体是通过一个抽象的FIFO队列来完成资源获取线程的排队工作，并通过一个int变量表示持有锁的状态
  • 锁是面向开发人员的，而同步器是JDK统一规范并简化了锁的实现，并抽象出来的公共基础部分，屏蔽了同步状态、同步队列的管理，和线程排队、通知、唤醒等机制

和AQS相关的类有：

  • ReentranLock
  • CountDownLatch
  • ReentrantReadWriteLock
  • Semaphore
  • 等

在这里插入图片描述

AQS 原理

  • 整体是通过一个抽象的FIFO队列来完成资源获取线程的排队工作，并通过一个int变量表示持有锁的状态
  • 如果共享资源被占用，就需要阻塞唤醒机制来保证锁的分配，这个机制主要是通过CLH队列的变体实现的，将暂时获取锁失败的线程，以及自身的等待状态封装成队列的节点对象node，放入队列中
  • 通过CAS、自旋等维护共享资源的状态，达到并发效果
  • 内部结构：
    • 队列的 头指针、尾指针
    • int 类型的同步状态的标识 state ，默认值0代表没有被占用，大于等于1代表被占用
    • 内部类node，将暂时获取锁失败的线程，以及自身的等待状态封装成队列的节点对象node
      • int类型变量 waitStatus 当前节点再队列中的等待状态，默认为0
        • 1表示线程被取消
        • -1表示后继线程需要被唤醒
        • -2表示等待conditon唤醒
        • -3表示共享式（锁分为共享和独占）同步状态获取将无条件地传播下去
      • 前一个节点的指针和后一个节点的指针
      • 请求线程

CLH队列

  • (Craig, Landin,Hagersten) 三个科学家名字的简称
  • 通过state状态判断是否阻塞，从尾部入队，头部出队

以ReentrantLock为例，加锁和解锁的步骤为：

lock

  • 当调用 lock 方法加锁时

    • 非公平锁，会先尝试通过cas 比较并交换的操作把 states 的状态值从 0更新为1，如果更新成功，就把持有锁的线程设置为自己
    • 更新失败就和公平锁一样，执行 AQS 的 acquire方法
    • 除此之外，公平和非公平锁的区别就是，再获取同步状态时，公平锁需要判断等待队列中再自己之前是否存在有效节点，如果有公平锁就需要排队
    • 因为公平锁讲究先到先得，线程再获取锁时，如果这个锁的等待队列已经有线程再等待，当前线程就会直接进入等待队列
    • 而非公平锁，不管是否有队列，如果可以获取锁，就会立刻占有锁的对象，所以第一个在队列里排队的线程苏醒后，仍然需要去竞争锁，且不一定能竞争到锁

acquire 方法源码：

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

acquire

  • 调用lock方法加锁，除非是非公平锁能直接拿到锁，其他情况下都是在调用acquire 方法
  • acquire 方法分为三种情况：
  • 调用 tryAcquire 方法尝试加锁；

    • AQS类的tryAcquire方法只是做了规范，方法内直接抛出异常，所以这个方法需要由子类去实现

    • 非公平锁的tryAcquire 方法会先判断锁的状态state是否为0，为0说明没有被其他线程占用，就立即使用cas操作变更state为1，变更成功就把持有锁的线程设置为自己，变更失败就表示加锁失败

    • 如果锁的状态为1，说明锁已经被占用，在比较当前线程和持有锁的线程是否一致，不一致就加锁失败

    • tryAcquire 方法公平和非公平锁的区别是

      • 再获取同步状态时，公平锁需要判断等待队列中再自己之前是否存在有效节点，如果有公平锁就需要排队
      • 非公平锁，不管是否有队列，如果可以获取锁，就会立刻占有锁的对象

    • 如果 tryAcquire 方法抢锁失败，就需要调用 addWaiter加入到等待队列

  • 加锁失败，调用 addWaite方法，进入等待队列；
    • acquire 方法的 addWaiter 方法创建的是独占的node节点，节点中封装的是当前线程
    • 首先要判断 链表的尾指针是否为空
      • 如果为空，就需要初始化链表，首先new一个空的哨兵节点，这个节点并不存储信息，只是作为占位使用，然后设置哨兵节点为头节点，然后把头节点赋值给尾节点
      • 当链表初始化完成后，或者链表中已经由其他节点时，就用CAS操作把新节点加入到链表尾部，如果节点加入链表失败就进行下一次循环，直到把节点加入成功为止
      • 如果不为空，直接用CAS操作把新节点加入到链表尾部，同样如果节点加入链表失败就进行循环，直到把节点加入成功为止
    • 节点成功入队后，需要调用acquireQueued 方法
  • 进入队列之后，调用acquireQueued 方法，线程进入阻塞状态，等待唤醒后才能继续执行
    • 首先获取当前节点的前置节点，如果前置节点是头节点，就尝试去获取锁
    • 如果获取锁成功，就把自己设为头节点，就把锁的state改为1，设置当前线程为持有锁的线程
    • 如果前置节点不是头节点，或者获取锁失败
      • 就需要判断前置节点的waitStatus状态值，waitStatus值默认为0，第一次进入循环，会把前置节点的waitStatus的值改为-1后，继续下一次循环后，会调用 LockSupport.park 方法阻塞当前线程，需要等待其他线程释放锁后，再唤醒阻塞的线程
      • 当持有锁的线程释放锁，且调用LockSupport.unpark 唤醒该线程后才能继续执行，LockSupport.unpark 唤醒的是头节点的下一个节点
      • 线程被唤醒后,检查线程是否被中断，如果线程没有被中断，就继续进行循环
      • 继续尝试去加锁，因为是非公平锁，所以有可能会加锁失败
        • 如果加锁成功，就把锁的state改为1，设置当前线程为持有锁的线程，并且把当前线程的节点设置为链表的头节点，原本的头节点会从链表中剔除
        • 因为每次唤醒的都是头节点的下一个节点，所以成功抢到到锁后，被唤醒的节点会成为新的头节点，后续会唤醒链表的下一个节点
    • 如果线程在等待过程中取消，没有获取到锁就跳出了循环，failed值为默认的true，就会执行cancelAcquire方法，取消正在排队的节点
      • 首先设置当前节点的线程为null，然后获取上一个没有取消的前置节点，
      • 把当前节点的 waitStatus 设置为1（1就是要取消的节点）
      • 如果当前节点是尾节点，就把上一个有效的节点设置为尾节点
      • 如果不是尾节点，并且满足出队条件，就变更链表中相关节点的前置和后置引用，剔除要取消的节点

非公平锁的 tryAcquire 方法源码

        final boolean nonfairTryAcquire(int acquires) {final Thread current = Thread.currentThread();int c = getState();//先判断锁的状态state是否为0，为0说明没有被其他线程占用if (c == 0) {//为0说明没有被其他线程占用，使用cas操作变更state为1if (compareAndSetState(0, acquires)) {//变更成功就把持有锁的线程设置为自己，变更失败就表示加锁失败setExclusiveOwnerThread(current);return true;}}//如果锁的状态为1，说明锁已经被占用，在比较当前线程和持有锁的线程是否一致else if (current == getExclusiveOwnerThread()) {int nextc = c + acquires;if (nextc < 0) // overflowthrow new Error("Maximum lock count exceeded");setState(nextc);return true;}//加锁失败return false;}

addWaiter(Node.EXCLUSIVE) 加入等待队列 源码：

  • Node.EXCLUSIVE 代表的是独占的节点，也就是排他锁
private Node addWaiter(Node mode) {//node节点中封装的是当前线程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(node);return node;
}private Node enq(final Node node) {for (;;) {//尾节点Node t = tail;//如果尾节点为nullif (t == null) { // Must initialize//就需要new一个node节点，并且设置为头节点，然后把头节点赋值给尾节点if (compareAndSetHead(new Node()))tail = head;} else {//当链表初始化完成后，或者链表中已经由其他节点时//把要加入链表的新节点的前指针设置为尾节点node.prev = t;//并且把新加入的节点设置为尾节点if (compareAndSetTail(t, node)) {//设置成功就之前尾节点的后指针指向新节点，这样新节点就变成了新的尾节点，如果设置失败，就继续循环，直到把新节点加入到链表尾部为止t.next = node;return t;}}}}

acquireQueued 源码

  • 线程进入阻塞状态，等待唤醒后才能继续执行
	//arg为1，独占锁final boolean acquireQueued(final Node node, int arg) {boolean failed = true;try {boolean interrupted = false;for (;;) {//获得node节点的前置节点final Node p = node.predecessor();//node节点的前置节点是否为头节点，如果是就尝试去获取锁if (p == head && tryAcquire(arg)) {setHead(node);p.next = null; // help GCfailed = false;return interrupted;}//判断node节点的前置节点的waitStatus状态，默认情况下都是0，在第二次循环的时候，就会改成-1，然后执行parkAndCheckInterrupt方法//parkAndCheckInterrupt方法会阻塞当前线程//也就是后面的节点会把前面节点的 waitStatus 改为-1if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())interrupted = true;}} finally {if (failed)cancelAcquire(node);}}/*** waitStatus 当前节点再队列中的等待状态默认为01表示线程获取锁的请求被取消-1表示线程已经准备好了-2表示节点在等待队列中，等待唤醒-3表示共享式（锁分为共享和独占）同步状态获取将无条件地传播下去*/private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {//前置节点的状态int ws = pred.waitStatus;// SIGNAL= -1 ， 当线程再次进行循环的时候，前一个节点的waitStatus已经被设置为-1，就返回trueif (ws == Node.SIGNAL)return true;//线程被取消if (ws > 0) {do {node.prev = pred = pred.prev;} while (pred.waitStatus > 0);pred.next = node;} else {//如果前置节点的 waitStatus不等于-1也不大于0，就把waitStatus的值改为-1后，返回falsecompareAndSetWaitStatus(pred, ws, Node.SIGNAL);}return false;}//阻塞当前线程private final boolean parkAndCheckInterrupt() {//验证当前线程的通行证，阻塞当前线程LockSupport.park(this);//被唤醒后,检查线程是否被中断，如果线程没有被中断，就返回 falsereturn Thread.interrupted();}

取消正在进行的获取尝试

	// node 为需要取消的节点private void cancelAcquire(Node node) {// Ignore if node doesn't existif (node == null)return;//设置当前节点的线程为nullnode.thread = null;//获取上一个节点Node pred = node.prev;//waitStatus > 0 ，表示上一个节点也要取消while (pred.waitStatus > 0)//那么就一直向上找，直到找到没有取消的前置节点node.prev = pred = pred.prev;//获取不会取消的前置节点的下一个节点Node predNext = pred.next;//把当前节点的 waitStatus 设置为1，1就是要取消的节点node.waitStatus = Node.CANCELLED;//如果当前节点是尾节点，就把上一个还有效的节点设置为尾节点if (node == tail && compareAndSetTail(node, pred)) {//设置成功，就把上一个节点的后置节点设置为null,这样上一个还有效的节点就成为了尾节点compareAndSetNext(pred, predNext, null);} else {//否则int ws;//前置节点不能是头节点，因为头节点只是占位节点，并且满足出队条件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 {unparkSuccessor(node);}node.next = node; // help GC}}
unlock

unlock 源码，其实是再调用release方法

    public void unlock() {sync.release(1);}

release方法会首先尝试释放锁

  • tryRelease 会把持有锁的线程为null，并且把锁的state设置为0
  • 如果链表被初始化过，有在等待的线程节点，头节点就不为空，且waitStatus值为-1
  • 接下来会把头节点的waitStatus的改为0，如果头节点的下一个节点不为null，就调用LockSupport.unpark 方法，唤醒头节点的下一个节点
    public final boolean release(int arg) {if (tryRelease(arg)) {Node h = head;if (h != null && h.waitStatus != 0)unparkSuccessor(h);return true;}return false;}

AQS的tryRelease方法，同样没有做实现，需要子类自己去实现，下面是ReentrantLock的实现

		protected final boolean tryRelease(int releases) {//传入的releases为1，持有锁的线程State为1，所以C为0int c = getState() - releases;//如果当前线程不等于持有锁的线程会抛出异常，这种情况一般不会出现if (Thread.currentThread() != getExclusiveOwnerThread())throw new IllegalMonitorStateException();boolean free = false;//c等于0，就设置持有锁的线程为null,并且把state设置为0，返回trueif (c == 0) {free = true;setExclusiveOwnerThread(null);}setState(c);return free;}

unparkSuccessor

    private void unparkSuccessor(Node node) {int ws = node.waitStatus;if (ws < 0)//重新把头节点的waitStatus值改为0compareAndSetWaitStatus(node, ws, 0);//头节点的下一个节点Node s = node.next;//如果链表被初始化过，有在等待的线程节点，头节点的后置节点就不为null//如果链表后面还有其他节点，那么头节点的后置节点waitStatus值就为-1if (s == null || s.waitStatus > 0) {s = null;for (Node t = tail; t != null && t != node; t = t.prev)if (t.waitStatus <= 0)s = t;}//如果头节点的下一个节点不为null，就直接调用 LockSupport.unpark 方法，唤醒头节点的下一个节点if (s != null)LockSupport.unpark(s.thread);}

这篇关于AQS 抽象队列同步器的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/580761。

相关文章

hdu1180（广搜+优先队列）

hdu1180（广搜+优先队列）

此题要求最少到达目标点T的最短时间，所以我选择了广度优先搜索，并且要用到优先队列。 另外此题注意点较多，比如说可以在某个点停留，我wa了好多两次，就是因为忽略了这一点，然后参考了大神的思想，然后经过反复修改才AC的 这是我的代码 #include<iostream>#include<algorithm>#include<string>#include<stack>#include<
阅读更多...
poj 3190 优先队列+贪心

poj 3190 优先队列+贪心

题意： 有n头牛，分别给他们挤奶的时间。 然后每头牛挤奶的时候都要在一个stall里面，并且每个stall每次只能占用一头牛。 问最少需要多少个stall，并输出每头牛所在的stall。 e.g 样例： INPUT: 51 102 43 65 84 7 OUTPUT: 412324 HINT: Explanation of the s
阅读更多...
poj 2431 poj 3253 优先队列的运用

poj 2431 poj 3253 优先队列的运用

poj 2431： 题意： 一条路起点为0， 终点为l。 卡车初始时在0点，并且有p升油，假设油箱无限大。 给n个加油站，每个加油站距离终点 l 距离为 x[i]，可以加的油量为fuel[i]。 问最少加几次油可以到达终点，若不能到达，输出-1。 解析： 《挑战程序设计竞赛》： “在卡车开往终点的途中，只有在加油站才可以加油。但是，如果认为“在到达加油站i时，就获得了一
阅读更多...
poj3750约瑟夫环，循环队列

poj3750约瑟夫环，循环队列

Description 有N个小孩围成一圈，给他们从1开始依次编号，现指定从第W个开始报数，报到第S个时，该小孩出列，然后从下一个小孩开始报数，仍是报到S个出列，如此重复下去，直到所有的小孩都出列（总人数不足S个时将循环报数），求小孩出列的顺序。 Input 第一行输入小孩的人数N（N<=64） 接下来每行输入一个小孩的名字(人名不超过15个字符) 最后一行输入W,S (W < N)，用
阅读更多...
POJ2010 贪心优先队列

POJ2010 贪心优先队列

c头牛,需要选n头（奇数）；学校总共有f的资金， 每头牛分数score和学费cost，问合法招生方案中,中间分数（即排名第(n+1)/2）最高的是多少。 n头牛按照先score后cost从小到大排序； 枚举中间score的牛,  预处理左边与右边的最小花费和。 预处理直接优先队列贪心 public class Main {public static voi
阅读更多...
Java并发编程之——BlockingQueue（队列）

Java并发编程之——BlockingQueue（队列）

一、什么是BlockingQueue BlockingQueue即阻塞队列，从阻塞这个词可以看出，在某些情况下对阻塞队列的访问可能会造成阻塞。被阻塞的情况主要有如下两种： 1. 当队列满了的时候进行入队列操作2. 当队列空了的时候进行出队列操作123 因此，当一个线程试图对一个已经满了的队列进行入队列操作时，它将会被阻塞，除非有另一个线程做了出队列操作；同样，当一个线程试图对一个空
阅读更多...
FreeRTOS学习笔记（六）队列

FreeRTOS学习笔记（六）队列

提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档 文章目录 前言一、队列的基本内容1.1 队列的引入1.2 FreeRTOS 队列的功能与作用1.3 队列的结构体1.4 队列的使用流程 二、相关API详解2.1 xQueueCreate2.2 xQueueSend2.3 xQueueReceive2.4 xQueueSendFromISR2.5 xQueueRecei
阅读更多...
多线程篇（阻塞队列- LinkedBlockingDeque）（持续更新迭代）

多线程篇（阻塞队列- LinkedBlockingDeque）（持续更新迭代）

目录 一、LinkedBlockingDeque是什么 二、核心属性详解 三、核心方法详解 addFirst(E e) offerFirst(E e) putFirst(E e) removeFirst() pollFirst() takeFirst() 其他 四、总结 一、LinkedBlockingDeque是什么 首先queue是一种数据结构，一个集合中
阅读更多...
工厂方法模式和抽象工厂模式的区别

工厂方法模式和抽象工厂模式的区别

区别  工厂方法模式： 一个抽象产品类，可以派生出多个具体产品类。    一个抽象工厂类，可以派生出多个具体工厂类。    每个具体工厂类只能创建一个具体产品类的实例。 抽象工厂模式： 多个抽象产品类，每个抽象产品类可以派生出多个具体产品类。    一个抽象工厂类，可以派生出多个具体工厂类。    每个具体工厂类可以创建多个具体产品类的实例。    区别： 工厂方法模式只有一个抽象产品类
阅读更多...
Java消息队列：RabbitMQ与Kafka的集成与应用

Java消息队列：RabbitMQ与Kafka的集成与应用

Java消息队列：RabbitMQ与Kafka的集成与应用 大家好，我是微赚淘客返利系统3.0的小编，是个冬天不穿秋裤，天冷也要风度的程序猿！ 在现代的分布式系统中，消息队列是实现系统间通信、解耦和提高可扩展性的重要组件。RabbitMQ和Kafka是两个广泛使用的消息队列系统，它们各有特点和优势。本文将介绍如何在Java应用中集成RabbitMQ和Kafka，并展示它们的应用场景。 消息队
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2