利用AQS(AbstractQueuedSynchronizer)实现一个线程同步器

简介: 利用AQS(AbstractQueuedSynchronizer)实现一个线程同步器


1. 前言

我们在上篇文章带领大家阅读了AQS的源码,理解了AQS的底层实现,AQS 为构建锁和同步器提供了一些通用功能的实现,今天我们就实践一下,基于AQS实现一个同步器,看看AQS如何帮助我们快速构建同步器。


2. 什么是同步器

同步器是一种用于线程同步的工具,控制多个线程对共享资源的并发访问。如Semaphore信号量,CountDownLatch (倒计时器)、CyclicBarrier(循环屏障)都属于同步器。

3. 同步器实现思路

在上一篇文章中我们就已经了解了AQS是基于 模版方法模式 的,他提供了一些抽象方法交付子类实现,我们基于此的自定义同步器设计一般需要如下两步:


1. 使用者继承 AbstractQueuedSynchronizer 并重写指定的方法;


2. 将 AQS 组合在自定义同步组件的实现中,并调用其模板方法,而这些模板方法会调用使用者重写的方法。


在模版方法模式下,有个很重要的东西,那就是“钩子方法”,这是一种抽象类中的方法,一般使用 protected 关键字修饰,可以给与默认实现,空方法居多,其内容逻辑由子类实现,为什么不适用抽象方法呢?因为,抽象方法需要子类全部实现,增加大量代码冗余!


钩子方法:由抽象类声明并且实现,子类也可以选择加以扩展。通常抽象类会给出一个空的钩子方法,也就是没有实现的扩展。区别在于抽象方法必须实现,而钩子方法可以不实现。也就是说钩子方法为你在实现某一个抽象类的时候提供了可选项,相当于预先提供了一个默认配置。


Ok,有了这层理论知识,我们去看看Java中根据AQS实现的同步工具类有哪些吧

Semaphore(信号量)

在前面我们讲过的synchronized 和 ReentrantLock 都是一次只允许一个线程访问某个资源,而Semaphore(信号量)可以用来控制同时访问特定资源的线程数量,它并不能保证线程安全。


我们下面来看一个关于Semaphore的使用示例:

public class Test {
    private final Semaphore semaphore;
 
    /**
     * 构造方法初始化信号量
     * @param limit
     */
    public Test(int limit) {
        this.semaphore = new Semaphore(limit);
    }
 
    public void useResource() {
        try {
            semaphore.acquire();
            // 使用资源
            System.out.println("资源use:" + Thread.currentThread().getName());
            Thread.sleep(1000); // 模拟资源使用时间
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            semaphore.release();
            System.out.println("资源release:" + Thread.currentThread().getName());
        }
    }
 
    public static void main(String[] args) {
        // 限制3个线程同时访问资源
        Test pool = new Test(3);
 
        for (int i = 0; i < 4; i++) {
            new Thread(pool::useResource).start();
        }
    }
}


输出

资源use:Thread-1
资源use:Thread-0
资源use:Thread-2
资源release:Thread-0
资源release:Thread-1
资源release:Thread-2
资源use:Thread-3
资源release:Thread-3

由此结果可看出,我们成功的将同时访问共享资源的线程数限制在了不超过3个的级别,这里面涉及到了Semaphore的两个主要方法:acquire()和release()


acquire():获取许可,跟进这个方法后,我们会发现其内部调用了AQS的一个final 方法acquireSharedInterruptibly(),这个方法中又调用了tryAcquireShared(arg)放,作为AQS中的钩子方法,这个方法的实现在Semaphore的两个静态内部类 FairSync(公平模式) 和 NonfairSync(非公平模式) 中。

release():释放许可,同样跟入这个方法,里面用了AQS的releaseShared(),而在这个方法内也毫无疑问的用了tryReleaseShared(int arg)这个钩子方法,原理同上,不再冗释。

以上Semaphore的具体原理实现可以参阅AQS同步器源码解读,这里就不展开说明了


4. 代码实现

有了上面的一系列学习,我们现在来手写一个自定义的同步器(我们这里写一个互斥锁,同一时刻只允许一个线程获取资源)吧,原理都是差不多的,开始前,先贴上AQS中的几个钩子方法,防止待会忘记。

//独占方式。尝试获取资源,成功则返回true,失败则返回false。
protected boolean tryAcquire(int)
//独占方式。尝试释放资源,成功则返回true,失败则返回false。
protected boolean tryRelease(int)
//共享方式。尝试获取资源。负数表示失败;0表示成功,但没有剩余可用资源;正数表示成功,且有剩余资源。
protected int tryAcquireShared(int)
//共享方式。尝试释放资源,成功则返回true,失败则返回false。
protected boolean tryReleaseShared(int)
//该线程是否正在独占资源。只有用到condition才需要去实现它。
protected boolean isHeldExclusively()

4.1. 创建互斥锁类

我们新建互斥锁类OnlySyncByAQS

/**
 * @author Luckysj @刘仕杰
 * @description 基于AQS实现的互斥锁类
 * @create 2024/04/20 20:44:07
 */
public class OnlySyncByAQS {
 
    
}


4.2  编写静态内部类,继承AQS


Semaphore控制内部的两个静态内部类来实现公平与非公平的资源获取,我们也可以学习Semaphore的设计,编写一个静态内部类Sync作为同步器,我们重写AQS的三个钩子函数tryAcquire(),tryRelease(),isHeldExclusively()

/**
 * @author Luckysj @刘仕杰
 * @description 基于AQS实现的互斥锁类
 * @create 2024/04/20 20:44:07
 */
public class OnlySyncByAQS {
/**
     * 静态内部类,继承AQS,重写钩子方法
     */
    private static class Sync extends AbstractQueuedSynchronizer {
 
       
    }
    
}

4.3  内部类实现AQS钩子函数

我们首先要重写tryAcquire(),这个方法可以以独占的方式获取同步状态。

 /**
         * 重写AQS的tryAcquire方法,独占方式,尝试获取资源。
         */
        @Override
        protected boolean tryAcquire(int arg) {
            //CAS 尝试更改状态
            if (compareAndSetState(0, 1)) {
                //独占模式下,设置锁的持有者为当前线程,来自于AOS
                setExclusiveOwnerThread(Thread.currentThread());
                System.out.println(Thread.currentThread().getName()+"获取锁成功");
                return true;
            }
            System.out.println(Thread.currentThread().getName()+"获取锁失败");
            return false;
        }


compareAndSetState这个方法由AQS提供,内部使用CAS操作修改state值,保证修改操作原子性

接着重写 tryRelease(),这个方法可以以独占的方式释放同步状态

 /**
         * 独占方式。尝试释放资源,成功则返回true,失败则返回false。
         * @param arg
         * @return
         */
        @Override
        protected boolean tryRelease(int arg) {
            if (getState() == 0) {
                throw new IllegalMonitorStateException();
            }
            //置空锁的持有者
            setExclusiveOwnerThread(null);
            //改状态为0,未锁定状态
            setState(0);
            System.out.println(Thread.currentThread().getName()+"释放锁成功!");
            return true;
        }

接着重写 isHeldExclusively方法,这个可以用来判断当前线程是否独占了同步状态。

         /**
         * 判断该线程是否正在独占资源,返回state=1
         * @return
         */
        @Override
        protected boolean isHeldExclusively() {
            return getState() == 1;
        }

4.3  封装lock,unlock方法

接下来我们在外部的互斥锁封装类中完成lock,unlock方法,对内部的Sync自定义同步器的使用进行封装,简化使用方式。

   private final Sync sync = new Sync();
 
    /**
     * 获取许可,给资源上锁
     */
    public void lock() {
        sync.acquire(1);
    }
 
    /**
     * 释放许可,解锁
     */
    public void unlock() {
        sync.release(1);
    }
 
    /**
     * 判断是否独占
     * @return
     */
    public boolean isLocked() {
        return sync.isHeldExclusively();
    }


4.4. 测试

新建测试类,来测试我们自定义的互斥锁同步器

public class Test {
 
    private OnlySyncByAQS onlySyncByAQS = new OnlySyncByAQS();
 
    public void use(){
        onlySyncByAQS.lock();
        try {
            //休眠1秒获取使用共享资源
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            onlySyncByAQS.unlock();
        }
    }
 
    public static void main(String[] args) throws InterruptedException {
        Test test = new Test();
        //多线程竞争资源,每次仅一个线程拿到锁
        for (int i = 0; i < 3; i++) {
            new Thread(()->{
                test.use();
            }).start();
        }
    }
}


运行结果如下,可以看到同一时间只有一个线程可以获取到锁:

5. 总结


本次我们学习了如何使用AQS实现一个互斥锁同步器,在实现过程中,我们主要就是操作state这个同步状态,线程通过修改state值为1来占用资源,如果修改失败,也就是获取锁失败,就会被加入到AQS的CLH双向队列中(可见上篇文章讲解AQS队列同步器源码解读阻塞等待。这样就实现了互斥锁,一个时间段只能有一个线程占用共享资源。


我们只需在钩子函数中对state的处理逻辑稍做修改,也可以实现共享锁,如Semaphore的实现,这就是 AQS 的强大所在。

相关文章
|
2月前
|
Java 开发者
解锁并发编程新姿势!深度揭秘AQS独占锁&ReentrantLock重入锁奥秘,Condition条件变量让你玩转线程协作,秒变并发大神!
【8月更文挑战第4天】AQS是Java并发编程的核心框架,为锁和同步器提供基础结构。ReentrantLock基于AQS实现可重入互斥锁,比`synchronized`更灵活,支持可中断锁获取及超时控制。通过维护计数器实现锁的重入性。Condition接口允许ReentrantLock创建多个条件变量,支持细粒度线程协作,超越了传统`wait`/`notify`机制,助力开发者构建高效可靠的并发应用。
77 0
|
2月前
|
Java 开发者
解锁Java并发编程的秘密武器!揭秘AQS,让你的代码从此告别‘锁’事烦恼,多线程同步不再是梦!
【8月更文挑战第25天】AbstractQueuedSynchronizer(AQS)是Java并发包中的核心组件,作为多种同步工具类(如ReentrantLock和CountDownLatch等)的基础。AQS通过维护一个表示同步状态的`state`变量和一个FIFO线程等待队列,提供了一种高效灵活的同步机制。它支持独占式和共享式两种资源访问模式。内部使用CLH锁队列管理等待线程,当线程尝试获取已持有的锁时,会被放入队列并阻塞,直至锁被释放。AQS的巧妙设计极大地丰富了Java并发编程的能力。
34 0
|
安全 Java
AQS唤醒线程的时候为什么从后向前遍历
AQS唤醒线程的时候为什么从后向前遍历
1016 0
AQS唤醒线程的时候为什么从后向前遍历
并发编程-17AQS同步组件之 Semaphore 控制并发线程数的信号量
并发编程-17AQS同步组件之 Semaphore 控制并发线程数的信号量
66 0
AQS源码探究_06 Conditon条件队列(await方法、线程入队与挂起、signal方法)
AQS源码探究_06 Conditon条件队列(await方法、线程入队与挂起、signal方法)
AQS源码探究_06 Conditon条件队列(await方法、线程入队与挂起、signal方法)
|
2月前
|
存储 监控 Java
Java多线程优化:提高线程池性能的技巧与实践
Java多线程优化:提高线程池性能的技巧与实践
64 1
|
6天前
|
数据采集 负载均衡 安全
LeetCode刷题 多线程编程九则 | 1188. 设计有限阻塞队列 1242. 多线程网页爬虫 1279. 红绿灯路口
本文提供了多个多线程编程问题的解决方案,包括设计有限阻塞队列、多线程网页爬虫、红绿灯路口等,每个问题都给出了至少一种实现方法,涵盖了互斥锁、条件变量、信号量等线程同步机制的使用。
LeetCode刷题 多线程编程九则 | 1188. 设计有限阻塞队列 1242. 多线程网页爬虫 1279. 红绿灯路口
|
14天前
|
Java Spring
spring多线程实现+合理设置最大线程数和核心线程数
本文介绍了手动设置线程池时的最大线程数和核心线程数配置方法,建议根据CPU核数及程序类型(CPU密集型或IO密集型)来合理设定。对于IO密集型,核心线程数设为CPU核数的两倍;CPU密集型则设为CPU核数加一。此外,还讨论了`maxPoolSize`、`keepAliveTime`、`allowCoreThreadTimeout`和`queueCapacity`等参数的设置策略,以确保线程池高效稳定运行。
77 10
spring多线程实现+合理设置最大线程数和核心线程数
|
22天前
|
Java 数据库 Android开发
一个Android App最少有几个线程?实现多线程的方式有哪些?
本文介绍了Android多线程编程的重要性及其实现方法,涵盖了基本概念、常见线程类型(如主线程、工作线程)以及多种多线程实现方式(如`Thread`、`HandlerThread`、`Executors`、Kotlin协程等)。通过合理的多线程管理,可大幅提升应用性能和用户体验。
38 15
一个Android App最少有几个线程?实现多线程的方式有哪些?
|
8天前
|
Python
5-5|python开启多线程入口必须在main,从python线程(而不是main线程)启动pyQt线程有什么坏处?...
5-5|python开启多线程入口必须在main,从python线程(而不是main线程)启动pyQt线程有什么坏处?...