首页 > 其他分享 >探索抽象同步队列 AQS

探索抽象同步队列 AQS

时间:2023-09-25 09:44:35浏览次数:43  
标签:AQS 队列 获取 state 线程 arg 抽象

by emanjusaka from https://www.emanjusaka.top/archives/8 彼岸花开可奈何
本文欢迎分享与聚合,全文转载请留下原文地址。

前言

AbstractQueuedSynchronizer抽象同步队列简称AQS,它是实现同步器的基础组件,并发包中锁的底层就是使用AQS实现的。大多数开发者可能永远不会直接使用AQS,但是知道其原理对于架构设计还是很有帮助的。AQS是Java中的一个抽象类,全称是AbstractQueuedSynchronizer,即抽象队列同步器。它定义了两种资源共享模式:独占式和共享式。独占式每次只能有一个线程持有锁,例如ReentrantLock实现的就是独占式的锁资源;共享式允许多个线程同时获取锁,并发访问共享资源,ReentrantReadWriteLock和CountDownLatch等就是实现的这种模式。AQS维护了一个volatile的state变量和一个FIFO(先进先出)的队列。其中state变量代表的是竞争资源标识,而队列代表的是竞争资源失败的线程排队时存放的容器 。

一、原理

AQS的核心思想是通过一个FIFO的队列来管理线程的等待和唤醒,同时维护了一个state变量来表示同步状态,可以通过getState、setState、compareAndSetState函数修改其值。当一个线程想要获取锁时,如果state为0,则表示该线程获取锁成功,否则表示该线程获取锁失败。它将被放入等待队列中,直到满足特定条件才能再次尝试获取。当一个线程释放锁时,如果state为1,则表示该线程释放锁成功,否则表示该线程释放锁失败。AQS通过CAS操作来实现加锁和解锁。

1.1 CLH队列

image

AQS中的CLH队列锁是CLH锁的一种变体,将自旋操作改成了阻塞线程操作。AQS 中的对 CLH 锁数据结构的改进主要包括三方面:扩展每个节点的状态、显式的维护前驱节点和后继节点以及诸如出队节点显式设为 null 等辅助 GC 的优化。

在 AQS(AbstractQueuedSynchronizer)中使用的 CLH 队列,head 指针和 tail 指针分别指向 CLH 队列中的两个关键节点。

  1. head 指针:head 指针指向 CLH 队列中的首个节点,该节点表示当前持有锁的线程。当一个线程成功地获取到锁时,它就成为了持有锁的线程,并且会将该信息记录在 head 指针所指向的节点中。
  2. tail 指针:tail 指针指向 CLH 队列中的最后一个节点,该节点表示队列中最后一个等待获取锁的线程。当一个线程尝试获取锁时,它会生成一个新的节点,并将其插入到 CLH 队列的尾部,然后成为 tail 指针所指向的节点。这样,tail 指针的作用是标记当前 CLH 队列中最后一个等待获取锁的线程。

通过 head 指针和 tail 指针,CLH 队列能够维护一种有序的等待队列结构,保证线程获取锁的顺序和互斥访问的正确性。当一个线程释放锁时,它会修改当前节点的状态,并唤醒后继节点上的线程,让后续的线程能够及时感知锁的释放,并争夺获取锁的机会。

1.2 线程同步

对于AQS来说,线程同步的关键是对状态值state进行操作。state为0时表示没有线程持有锁,大于0时表示有线程持有锁。根据state是否属于一个线程,操作state的方式分为独占方式和共享方式。

在独占方式下获取和释放资源使用的方法为:

  • void acquire(int arg)
  • void acquireInterruptibly(int arg)
  • boolean release(int arg)

使用独占方式获取的资源是与具体线程绑定的,就是说如果一个线程获取到了资源,就会标记是这个线程获取到了,其他线程再尝试操作state获取资源时会发现当前该资源不是自己持有的,就会在获取失败后被阻塞。

在共享方式下获取和释放资源的方法为:

  • void acquireShared(int arg)
  • voidacquireSharedInterruptibly(int arg)
  • boolean releaseShared(int arg)。

对应共享方式的资源与具体线程是不相关的,当多个线程去请求资源时通过CAS方式竞争获取资源,当一个线程获取到了资源后,另外一个线程再次去获取时如果当前资源还能满足它的需要,则当前线程只需要使用CAS方式进行获取即可。

二、资源获取与释放

2.1 独占式

  1. 当一个线程调用acquire(int arg)方法获取独占资源时,会首先使用tryAcquire方法尝试获取资源,具体是设置状态变量state的值,成功则直接返回,失败则将当前线程封装为类型为Node.EXCLUSIVE的Node节点后插入到AQS阻塞队列的尾部,并调用LockSupport.park(this)方法挂起自己。

        public final void acquire(int arg) {
          if (! tryAcquire(arg) &&
              acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
              selfInterrupt();
       }
    
  1. 当一个线程调用release(int arg)方法时会尝试使用tryRelease操作释放资源,这里是设置状态变量state的值,然后调用LockSupport.unpark(thread)方法激活AQS队列里面被阻塞的一个线程(thread)。被激活的线程则使用tryAcquire尝试,看当前状态变量state的值是否能满足自己的需要,满足则该线程被激活,然后继续向下运行,否则还是会被放入AQS队列并被挂起。

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

2.2 共享式

  1. 当线程调用acquireShared(int arg)获取共享资源时,会首先使用tryAcquireShared尝试获取资源,具体是设置状态变量state的值,成功则直接返回,失败则将当前线程封装为类型为Node.SHARED的Node节点后插入到AQS阻塞队列的尾部,并使用LockSupport.park(this)方法挂起自己。

        public final void acquireShared(int arg) {
            if (tryAcquireShared(arg) < 0)
                doAcquireShared(arg);
        }
    
  1. 当一个线程调用releaseShared(int arg)时会尝试使用tryReleaseShared操作释放资源,这里是设置状态变量state的值,然后使用LockSupport.unpark(thread)激活AQS队列里面被阻塞的一个线程(thread)。被激活的线程则使用tryReleaseShared查看当前状态变量state的值是否能满足自己的需要,满足则该线程被激活,然后继续向下运行,否则还是会被放入AQS队列并被挂起。

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

三、基于AQS实现自定义同步器

基于AQS实现一个不可重入的独占锁,自定义AQS需要重写一系列函数,还需要定义原子变量state的含义。这里定义,state为0表示目前锁没有被线程持有,state为1表示锁已经被某一个线程持有,由于是不可重入锁,所以不需要记录持有锁的线程获取锁的次数。

package top.emanjusaka;

import java.io.Serializable;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

public class UnReentrantLock implements Lock, Serializable {
    // 借助 AbstractQueuedSynchronizer 实现
    private static class Sync extends AbstractQueuedSynchronizer {
        // 查看是否有线程持有锁
        protected boolean isHeldExclusively() {
            return getState() == 1;
        }
        // 尝试获取锁
        public boolean tryAcquire(int acquires) {
            assert acquires == 1;
            // 使用CAS 设置state
            if (compareAndSetState(0, 1)) {
                // 如果 CAS 操作成功,表示成功获得了锁。这时,通过 setExclusiveOwnerThread 方法将当前线程设置为独占锁的拥有者
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            // 如果 CAS 操作失败,即无法将 state 的值从 0 设置为 1,表示锁已被其他线程占用,无法获取锁,于是返回 false。
            return false;
        }
        // 尝试释放锁
        protected boolean tryRelease(int releases) {
            assert releases == 1;
            if (getState() == 0)
                throw new IllegalMonitorStateException();
            // 释放成功,将独占锁的拥有者设为null
            setExclusiveOwnerThread(null);
            // 将state的值设为0
            setState(0);
            return true;
        }

        Condition newCondition() {
            return new ConditionObject();
        }
    }

    private final Sync sync = new Sync();

    @Override
    public void lock() {
        sync.acquire(1);
    }

    @Override
    public boolean tryLock() {
        return sync.tryAcquire(1);
    }

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

    @Override
    public Condition newCondition() {
        return sync.newCondition();
    }

    public boolean isLocked() {
        return sync.isHeldExclusively();
    }

    public void lockInterruptibly() throws InterruptedException {
        sync.acquireInterruptibly(1);
    }

    public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException {
        return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
    }

}

在释放锁时并没有使用 CAS(Compare and Swap)操作,而是直接使用了 setState​ 方法来将 state​ 的值设置为 0。

这是因为在释放锁的过程中,并不需要涉及到多线程并发的问题。只有持有锁的线程才能够释放锁,其他线程无法对锁进行操作。因此,不需要使用 CAS 来进行原子性的状态更新。

在这种情况下,可以直接使用普通的方法来设置 state​ 的值为 0,将独占锁的拥有者设为 null。因为只有一个线程可以操作这个锁,不存在并发竞争的情况,也就不需要使用 CAS 来保证原子性。

需要注意的是,当调用 tryRelease​ 方法时,应该保证当前线程是持有锁的线程,否则会抛出 IllegalMonitorStateException​ 异常。这是为了确保只有拥有锁的线程才能释放锁,防止误释放其他线程的锁。

四、参考资料

  1. 《并发编程之美》
  2. ​AbstractQueuedSynchronizer​​抽象类的源码

本文原创,才疏学浅,如有纰漏,欢迎指正。尊贵的朋友,如果本文对您有所帮助,欢迎点赞,并期待您的反馈,以便于不断优化。
原文地址: https://www.emanjusaka.top/archives/8
微信公众号:emanjusaka的编程栈

标签:AQS,队列,获取,state,线程,arg,抽象
From: https://www.cnblogs.com/emanjusaka/p/page_8.html

相关文章

  • 22消息队列实现进程间的通讯
    通过消息队列实现进程间的通讯 frommultiprocessingimportProcess,Queuefromtimeimportsleep#向队列中写入数据defwrite_task(q):ifnotq.full():foriinrange(5):message='消息'+str(i)q.put(message)......
  • 21python实现简单的消息队列
      frommultiprocessingimportQueue'''q=Queue(num)若括号中没有指定最大可接收的消息数量,或数量为负值,那么就代表可接收的消息数量没有上限(直到内存的尽头)。函数也是队列的初始化。Queue.qsize()返回当前队列包含的消息数量。Queue.empty()如果队列为空,返回T......
  • 【POJ 3253】Fence Repair 题解(贪心算法+优先队列+哈夫曼树)
    农夫约翰想修理牧场周围的一小段围栏。他测量了围栏,发现他需要N(1≤N≤20000)块木板,每块木板都有一定的整数长度Li(1≤Li≤50000)单位。然后,他购买了一块长度刚好足以锯入N块木板的长木板(即,其长度为Li长度的总和)。FJ忽略了“切口”,即锯切时锯屑损失的额外长度;你也应该忽略它。FJ伤心地......
  • 消息队列中,如何保证消息的顺序性?
    本文选自:advanced-java作者:yanglbme问:如何保证消息的顺序性?面试官心理分析其实这个也是用MQ的时候必问的话题,第一看看你了不了解顺序这个事儿?第二看看你有没有办法保证消息是有顺序的?这是生产系统中常见的问题。面试题剖析我举个例子,我们以前做过一个mysqlbinlog同步的系统,压......
  • 消息队列中,如何保证消息的顺序性?
    本文选自:advanced-java作者:yanglbme问:如何保证消息的顺序性?面试官心理分析其实这个也是用MQ的时候必问的话题,第一看看你了不了解顺序这个事儿?第二看看你有没有办法保证消息是有顺序的?这是生产系统中常见的问题。面试题剖析我举个例子,我们以前做过一个mysqlbinlog同步......
  • 【Java 基础篇】深入理解 Java 中的抽象类:提高代码可维护性与扩展性
    抽象类(AbstractClass)是Java面向对象编程中的一个重要概念。它允许我们定义一组抽象方法,这些方法可以被子类(类)实现。抽象类通常用于定义一些通用的方法和属性,但不能被实例化。本篇博客将深入探讨Java中抽象类的概念、语法和实际应用,适用于初学者,帮助你轻松理解和应用抽象类。什......
  • redis消息队列——发布订阅
    一、相关依赖<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId></dependency><dependency><grou......
  • 【POJ 1521】Entropy 题解(贪心算法+优先队列+哈夫曼树)
    熵编码器是一种数据编码方法,通过对删除了“浪费”或“额外”信息的消息进行编码来实现无损数据压缩。换句话说,熵编码去除了最初不需要的信息,以准确编码消息。高度的熵意味着一条消息包含大量浪费的信息;以ASCII编码的英文文本是具有极高熵的消息类型的示例。已经压缩的消息,如JPEG图......
  • Spring Boot中的消息队列集成
    介绍在现代应用程序中,消息队列已经成为了一种非常流行的解决方案,它可以帮助我们实现异步通信、解耦和扩展性。SpringBoot提供了对多种消息队列的集成支持,包括RabbitMQ、Kafka、ActiveMQ等。在本文中,我们将深入探讨SpringBoot中的消息队列集成。RabbitMQ集成RabbitMQ是一个流行......
  • JS实现任务队列
    引言假设有这么一个场景:前端订阅后台数据的变化,如果发生变化,则触发订阅回调;回调函数中,会执行一些耗时操作,如:请求接口,发送短信,存历史数据等;要求以上所有的操作都必须按照订阅触发的顺序执行;我们都知道,回调本身就是一种异步操作,我们仅仅依靠订阅回调无法保证回调中任务执行顺......