当前位置： 首页 > article >正文

AbstractQueuedSynchronizer

article 2024/10/31 15:13:49

AbstractQueuedSynchronizer

队列同步器AbstractQueuedSynchronizer（以下简称同步器或AQS），是用来构建锁或者其他同步组件的基础框架，它使用了一个int成员变量表示同步状态，通过内置的FIFO队列来完成资源获取线程的排队工作。并发包的大师（Doug Lea）期望它能够成为实现大部分同步需求的基础。

AQS使用方式和其中的设计模式

AQS的主要使用方式是继承，子类通过继承AQS并实现它的抽象方法来管理同步状态，在AQS里由一个int型的state来代表这个状态，在抽象方法的实现过程中免不了要对同步状态进行更改，这时就需要使用同步器提供的3个方法（getState()、setState(int newState)和compareAndSetState(int expect,int update)）来进行操作，因为它们能够保证状态的改变是安全的。

    /**
     * The synchronization state.
     */
    private volatile int state;

在实现上，子类推荐被定义为自定义同步组件的静态内部类，AQS自身没有实现任何同步接口，它仅仅是定义了若干同步状态获取和释放的方法来供自定义同步组件使用，同步器既可以支持独占式地获取同步状态，也可以支持共享式地获取同步状态，这样就可以方便实现不同类型的同步组件（ReentrantLock、ReentrantReadWriteLock和CountDownLatch等）。

同步器是实现锁（也可以是任意同步组件）的关键，在锁的实现中聚合同步器。可以这样理解二者之间的关系：

锁是面向使用者的，它定义了使用者与锁交互的接口（比如可以允许两个线程并行访问），隐藏了实现细节；

同步器面向的是锁的实现者，它简化了锁的实现方式，屏蔽了同步状态管理、线程的排队、等待与唤醒等底层操作。锁和同步器很好地隔离了使用者和实现者所需关注的领域。

实现者需要继承同步器并重写指定的方法，随后将同步器组合在自定义同步组件的实现中，并调用同步器提供的模板方法，而这些模板方法将会调用使用者重写的方法。

模板方法模式

同步器的设计基于模板方法模式。模板方法模式的意图是，定义一个操作中的算法的骨架，而将一些步骤的实现延迟到子类中。模板方法使得子类可以不改变一个算法的结构即可重定义该算法的某些特定步骤。我们最常见的就是Spring框架里的各种Template。

实际例子

我们开了个蛋糕店，蛋糕店不能只卖一种蛋糕呀，于是我们决定先卖奶油蛋糕，芝士蛋糕和慕斯蛋糕。三种蛋糕在制作方式上一样，都包括造型，烘焙和涂抹蛋糕上的东西。所以可以定义一个抽象蛋糕模型

/**
 * 类说明：抽象蛋糕模型
 */
public abstract class AbstractCake {
    protected abstract void shape();
    protected abstract void apply();
    protected abstract void brake();

    /*模板方法*/
    public final void run(){
        this.shape();
        this.apply();
        this.brake();
    }
}

然后就可以批量生产三种蛋糕

/**
 * 类说明：芝士蛋糕
 */
public class CheeseCake  extends AbstractCake {

    @Override
    protected void shape() {
        System.out.println("芝士蛋糕造型");
    }

    @Override
    protected void apply() {
        System.out.println("芝士蛋糕涂抹");
    }

    @Override
    protected void brake() {
        System.out.println("芝士蛋糕烘焙");
    }
}

public class CreamCake extends AbstractCake {
    
public class MouseCake extends AbstractCake {
    
AbstractCake cake = new CheeseCake();
AbstractCake cake2 = new CreamCake();
//AbstractCake cake3 = new MouseCake();
cake.run();

这样一来，不但可以批量生产三种蛋糕，而且如果日后有扩展，只需要继承抽象蛋糕方法就可以了，十分方便，我们天天生意做得越来越赚钱。突然有一天，我们发现市面有一种最简单的小蛋糕销量很好，这种蛋糕就是简单烘烤成型就可以卖，并不需要涂抹什么食材，由于制作简单销售量大，这个品种也很赚钱，于是我们也想要生产这种蛋糕。但是我们发现了一个问题，抽象蛋糕是定义了抽象的涂抹方法的，也就是说扩展的这种蛋糕是必须要实现涂抹方法。怎么办？我们可以将原来的模板修改为带钩子的模板。

/*模板方法*/
public final void run(){
    this.shape();
    if(this.shouldApply()){
        this.apply();
    }
    this.brake();
}

protected boolean shouldApply(){
    return true;
}

做小蛋糕的时候通过flag来控制是否涂抹，其余已有的蛋糕制作不需要任何修改可以照常进行。

/**
 * 类说明：小蛋糕
 */
public class SmallCake extends AbstractCake {

    private boolean flag = false;
    public void setFlag(boolean shouldApply){
        flag = shouldApply;
    }
    @Override
    protected boolean shouldApply() {
        return this.flag;
    }
    @Override
    protected void shape() {
        System.out.println("小蛋糕造型");
    }

    @Override
    protected void apply() {
        System.out.println("小蛋糕涂抹");
    }

    @Override
    protected void brake() {
        System.out.println("小蛋糕烘焙");
    }

}

AQS中的方法

模板方法
实现自定义同步组件时，将会调用同步器提供的模板方法
在这里插入图片描述

这些模板方法同步器提供的模板方法基本上分为3类：独占式获取与释放同步状态、共享式获取与释放、同步状态和查询同步队列中的等待线程情况。

可重写的方法

访问或修改同步状态的方法
重写同步器指定的方法时，需要使用同步器提供的如下3个方法来访问或修改同步状态。
•getState()：获取当前同步状态。
•setState(int newState)：设置当前同步状态。
•compareAndSetState(int expect,int update)：使用CAS设置当前状态，该方法能够保证状态设置的原子性。

AQS的基本思想CLH队列锁

CLH队列锁即Craig, Landin, and Hagersten (CLH) locks。
CLH队列锁也是一种基于链表的可扩展、高性能、公平的自旋锁，申请线程仅仅在本地变量上自旋，它不断轮询前驱的状态，假设发现前驱释放了锁就结束自旋。
当一个线程需要获取锁时：

  1. 创建一个的QNode，将其中的locked设置为true表示需要获取锁，myPred表示对其前驱结点的引用

  2. 线程A对tail域调用getAndSet方法，使自己成为队列的尾部，同时获取一个指向其前驱结点的引用myPred

线程B需要获得锁，同样的流程再来一遍

3.线程就在前驱结点的locked字段上旋转，直到前驱结点释放锁(前驱节点的锁值 locked == false)

4.当一个线程需要释放锁时，将当前结点的locked域设置为false，同时回收前驱结点

如上图所示，前驱结点释放锁，线程A的myPred所指向的前驱结点的locked字段变为false，线程A就可以获取到锁。
CLH队列锁的优点是空间复杂度低（如果有n个线程，L个锁，每个线程每次只获取一个锁，那么需要的存储空间是O（L+n），n个线程有n个myNode，L个锁有L个tail）。CLH队列锁常用在SMP体系结构下。
Java中的AQS是CLH队列锁的一种变体实现。
实现我们自己独占锁,不可重入demo

SelfLock

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 SelfLock implements Lock {
    // 静态内部类，自定义同步器
    private static class Sync extends AbstractQueuedSynchronizer {

        /*判断处于占用状态*/
        @Override
        protected boolean isHeldExclusively() {
            return getState()==1;
        }

        /*获得锁*/
        @Override
        protected boolean tryAcquire(int arg) {
            if(compareAndSetState(0,1)){
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        /*释放锁*/
        @Override
        protected boolean tryRelease(int arg) {
            if(getState()==0){
                throw new IllegalMonitorStateException();
            }
            setExclusiveOwnerThread(null);
            setState(0);
            //compareAndSetState(1,0);
            return true;
        }

        // 返回一个Condition，每个condition都包含了一个condition队列
        Condition newCondition() {
            return new ConditionObject();
        }
    }

    // 仅需要将操作代理到Sync上即可
    private final Sync sync = new Sync();

    public void lock() {
        System.out.println(Thread.currentThread().getName()+" ready get lock");
        sync.acquire(1);
        System.out.println(Thread.currentThread().getName()+" already got lock");
    }

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

    public void unlock() {
        System.out.println(Thread.currentThread().getName()+" ready release lock");
        sync.release(1);
        System.out.println(Thread.currentThread().getName()+" already released lock");
    }

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

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

    public boolean hasQueuedThreads() {
        return sync.hasQueuedThreads();
    }

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

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

SleepTools

import java.util.concurrent.TimeUnit;

/**
 * 
 *类说明：线程休眠辅助工具类
 */
public class SleepTools {
    
    /**
     * 按秒休眠
     * @param seconds 秒数
     */
    public static final void second(int seconds) {
        try {
            TimeUnit.SECONDS.sleep(seconds);
        } catch (InterruptedException e) {
        }
    }
    
    /**
     * 按毫秒数休眠
     * @param seconds 毫秒数
     */
    public static final void ms(int seconds) {
        try {
            TimeUnit.MILLISECONDS.sleep(seconds);
        } catch (InterruptedException e) {
        }
    }
}

TestSelfLock

import java.util.concurrent.locks.Lock;

public class TestMyLock {

    public void test() {
        final Lock lock = new SelfLock();
        class Worker extends Thread {

          public void run() {
                lock.lock();
                System.out.println(Thread.currentThread().getName());
                try {
                    SleepTools.second(1);
                } finally {
                    lock.unlock();
                }
            }
        }
        // 启动4个子线程
        for (int i = 0; i < 4; i++) {
            Worker w = new Worker();
            //w.setDaemon(true);
            w.start();
        }
        // 主线程每隔1秒换行
        for (int i = 0; i < 10; i++) {
            SleepTools.second(1);
            //System.out.println();
        }
    }

    public static void main(String[] args) {
        TestMyLock testMyLock = new TestMyLock();
        testMyLock.test();
    }
}

打印结果为：

Thread-0 ready get lock
Thread-2 ready get lock
Thread-3 ready get lock
Thread-1 ready get lock
Thread-0 already got lock
Thread-0
Thread-0 ready release lock
Thread-0 already released lock
Thread-2 already got lock
Thread-2
Thread-2 ready release lock
Thread-2 already released lock
Thread-3 already got lock
Thread-3
Thread-3 ready release lock
Thread-3 already released lock
Thread-1 already got lock
Thread-1
Thread-1 ready release lock
Thread-1 already released lock

Process finished with exit code 0

ReentrantLock的实现

锁的可重入

重进入是指任意线程在获取到锁之后能够再次获取该锁而不会被锁所阻塞，该特性的实现需要解决以下两个问题。

1）线程再次获取锁。锁需要去识别获取锁的线程是否为当前占据锁的线程，如果是，则再次成功获取。
2）锁的最终释放。线程重复n次获取了锁，随后在第n次释放该锁后，其他线程能够获取到该锁。锁的最终释放要求锁对于获取进行计数自增，计数表示当前锁被重复获取的次数，而锁被释放时，计数自减，当计数等于0时表示锁已经成功释放。

nonfairTryAcquire方法增加了再次获取同步状态的处理逻辑：通过判断当前线程是否为获取锁的线程来决定获取操作是否成功，如果是获取锁的线程再次请求，则将同步状态值进行增加并返回true，表示获取同步状态成功。同步状态表示锁被一个线程重复获取的次数。

如果该锁被获取了n次，那么前(n-1)次tryRelease(int releases)方法必须返回false，而只有同步状态完全释放了，才能返回true。可以看到，该方法将同步状态是否为0作为最终释放的条件，当同步状态为0时，将占有线程设置为null，并返回true，表示释放成功。

公平和非公平锁

ReentrantLock的构造函数中，默认的无参构造函数将会把Sync对象创建为NonfairSync对象，这是一个“非公平锁”；而另一个构造函数ReentrantLock(boolean fair)传入参数为true时将会把Sync对象创建为“公平锁”FairSync。

nonfairTryAcquire(int acquires)方法，对于非公平锁，只要CAS设置同步状态成功，则表示当前线程获取了锁，而公平锁则不同。tryAcquire方法，该方法与nonfairTryAcquire(int acquires)比较，唯一不同的位置为判断条件多了hasQueuedPredecessors()方法，即加入了同步队列中当前节点是否有前驱节点的判断，如果该方法返回true，则表示有线程比当前线程更早地请求获取锁，因此需要等待前驱线程获取并释放锁之后才能继续获取锁。

实现我们自己独占锁,可重入
ReenterSelfLock

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 ReenterSelfLock implements Lock {
    // 静态内部类，自定义同步器
    private static class Sync extends AbstractQueuedSynchronizer {

        // 是否处于占用状态
        protected boolean isHeldExclusively() {
            return getState() > 0;
        }

        // 当状态为0的时候获取锁
        public boolean tryAcquire(int acquires) {
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }else if(getExclusiveOwnerThread()==Thread.currentThread()){
                setState(getState()+1);
                return  true;
            }
            return false;
        }

        // 释放锁，将状态设置为0
        protected boolean tryRelease(int releases) {
            if(getExclusiveOwnerThread()!=Thread.currentThread()){
                throw new IllegalMonitorStateException();
            }
            if (getState() == 0)
                throw new IllegalMonitorStateException();

            setState(getState()-1);
            if(getState()==0){
                setExclusiveOwnerThread(null);
            }
            return true;
        }

        // 返回一个Condition，每个condition都包含了一个condition队列
        Condition newCondition() {
            return new ConditionObject();
        }
    }

    // 仅需要将操作代理到Sync上即可
    private final Sync sync = new Sync();

    public void lock() {
        System.out.println(Thread.currentThread().getName()+" ready get lock");
        sync.acquire(1);
        System.out.println(Thread.currentThread().getName()+" already got lock");
    }

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

    public void unlock() {
        System.out.println(Thread.currentThread().getName()+" ready release lock");
        sync.release(1);
        System.out.println(Thread.currentThread().getName()+" already released lock");
    }

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

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

    public boolean hasQueuedThreads() {
        return sync.hasQueuedThreads();
    }

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

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

TestReenterSelfLock

import java.util.concurrent.locks.Lock;

/**
 *类说明：
 */
public class TestReenterSelfLock {

    static final Lock lock = new ReenterSelfLock();

    public void reenter(int x){
        lock.lock();
        try {
            System.out.println(Thread.currentThread().getName()+":递归层级:"+x);
            int y = x - 1;
            if (y==0) return;
            else{
                reenter(y);
            }
        } finally {
            lock.unlock();
        }

    }

    public void test() {
        class Worker extends Thread {
          public void run() {
                System.out.println(Thread.currentThread().getName());
                SleepTools.second(1);
                reenter(3);
            }
        }
        // 启动3个子线程
        for (int i = 0; i < 3; i++) {
            Worker w = new Worker();
            w.start();
        }
        // 主线程每隔1秒换行
        for (int i = 0; i < 100; i++) {
            SleepTools.second(1);
        }
    }

    public static void main(String[] args) {
        TestReenterSelfLock testMyLock = new TestReenterSelfLock();
        testMyLock.test();
    }
}

打印结果为：

Thread-0
Thread-2
Thread-1
Thread-1 ready get lock
Thread-0 ready get lock
Thread-2 ready get lock
Thread-1 already got lock
Thread-1:递归层级:3
Thread-1 ready get lock
Thread-1 already got lock
Thread-1:递归层级:2
Thread-1 ready get lock
Thread-1 already got lock
Thread-1:递归层级:1
Thread-1 ready release lock
Thread-1 already released lock
Thread-1 ready release lock
Thread-1 already released lock
Thread-1 ready release lock
Thread-1 already released lock
Thread-0 already got lock
Thread-0:递归层级:3
Thread-0 ready get lock
Thread-0 already got lock
Thread-0:递归层级:2
Thread-0 ready get lock
Thread-0 already got lock
Thread-0:递归层级:1
Thread-0 ready release lock
Thread-0 already released lock
Thread-0 ready release lock
Thread-0 already released lock
Thread-0 ready release lock
Thread-0 already released lock
Thread-2 already got lock
Thread-2:递归层级:3
Thread-2 ready get lock
Thread-2 already got lock
Thread-2:递归层级:2
Thread-2 ready get lock
Thread-2 already got lock
Thread-2:递归层级:1
Thread-2 ready release lock
Thread-2 already released lock
Thread-2 ready release lock
Thread-2 already released lock
Thread-2 ready release lock
Thread-2 already released lock

Process finished with exit code 0

http://www.kler.cn/a/373135.html

相关文章：

  • C#核心（5）构造，析构，垃圾回收
  • 爬虫+数据保存2
  • 模块化沙箱是什么？如何运用到源代码保密中？
  • 状态机模型
  • ubuntu中安装使用Geph
  • 【Stable Diffusion - Ai】小白入门必看（涂鸦、涂鸦重绘、局部重绘和重绘蒙版篇）！真材实料！不卖课！！！
  • 0基础入门matlab
  • K8s 容器的定向调度与亲和性
  • 大厂面经：京东嵌入式面试题及参考答案
  • 安全研究 | 不同编程语言中 IP 地址分类的不一致性
  • ZeroNL2SQL：零样本 NL2SQL
  • 第三百零三节 Log4j教程 - Log4j安装
  • MATLAB/Simulink学习|在Simulink中调用C语言-02使用C Function 实现积分运算
  • orbslam安装
  • Unity 打包AB Timeline 引用丢失，错误问题
  • 从零开始机器学习——基于PyTorch构建你的第一个线性回归模型
  • VS离线安装NuGet包
  • WordPress插件 Lightsns主题专版-AI内容生成 V1.6 AI驱动的内容创作工具
  • 基于深度学习的声纹识别
  • ubuntu限制网速方法
  • 部署通义千问到后端-过程记录
  • 阿里云开源 AI 应用开发框架：Spring AI Alibaba
  • ip命令网络配置详解
  • GIS基础知识：WKT格式、WKB格式
  • 「C/C++」C++标准库之#include＜fstream＞文件流
  • 深入布局- grid布局