JavaSE中AQS详解

た入场券 2023-10-13 17:48 54阅读 0赞

注意：本文属于Java进阶知识

**目录**

一、什么是AQS

二、AQS底层实现原理

三、基于CAS(Compare And Swap)+LockSupport+AQS(AbstractQueueSynchronizer)写自己的锁

四、Semaphore(信号量)及其API

五、CountDownLatch及其API

--------------------

## 一、什么是AQS ##

> 在Java中，**AQS是指AbstractQueuedSynchronizer**，是Java并发编程中的一个关键类。AQS是用来构建同步器（Synchronizer）的框架，它提供了一种实现线程安全性和管理等待线程的机制。
> 
> **AQS基于一个FIFO（先进先出）的等待队列**，可以被子类扩展来实现各种同步器，如ReentrantLock、CountDownLatch、Semaphore等。AQS通过定义了若干个方法来管理线程的获取和释放资源，其中最为重要的方法是acquire()和release()。
> 
> 在使用AQS时，子类需要重写AQS的一些方法来实现自定义的同步策略。AQS提供了一个内部状态变量和线程队列，通过原子操作来维护状态和等待线程。当某个线程需要获取同步资源时，如果资源已经被其他线程占用，则该线程会被加入等待队列并进入阻塞状态。当资源释放时，AQS会自动唤醒等待队列中的线程，使其有机会再次尝试获取资源。
> 
> AQS是Java并发编程中非常重要的一个组件，它为实现自定义的同步器提供了强大的支持，并且在Java的并发框架（如并发集合类、线程池等）的实现中得到广泛使用。

--------------------

## 二、AQS底层实现原理 ##

> AQS的底层源码非常复杂，涉及到许多细节和高级概念。以下是对AQS源码的一个概要介绍：
> 
> 1. 内部状态变量：AQS内部使用一个state变量来表示同步状态。子类可以使用该变量来记录自定义的同步状态信息。
> 
> 2. 线程队列：AQS维护了一个双向链表的线程队列，用于存储等待获取同步资源的线程。线程队列采用先进先出（FIFO）的顺序，保证等待时间最长的线程优先获取资源。
> 
> 3. Node节点：线程队列中的每个线程都被封装成一个Node节点，其中包含了线程本身的信息以及一些标记位。Node节点使用了prev和next指针来维护链表结构。
> 
> 4. acquire()方法：当线程需要获取同步资源时，它会调用acquire()方法。在该方法内部，AQS首先会尝试直接获取资源，如果获取成功则线程可以继续执行。如果获取失败，则线程会被包装成一个Node节点并入队，同时进入自旋或阻塞状态，等待资源的释放。
> 
> 5. release()方法：当线程释放同步资源时，它会调用release()方法。在该方法内部，AQS会将资源的状态更新，并检查是否有等待的线程需要被唤醒。如果存在等待的线程，AQS会从等待队列中取出线程并唤醒它们。
> 
> 6. 其他方法：除了acquire()和release()方法外，AQS还提供了一些其他方法，如tryAcquire()和tryRelease()等。这些方法允许子类自定义获取和释放资源的机制。

--------------------

## 三、基于CAS(Compare And Swap)+LockSupport+AQS(AbstractQueueSynchronizer)写自己的锁 ##

package JUC;
    
    import java.util.concurrent.ConcurrentLinkedDeque;
    import java.util.concurrent.atomic.AtomicInteger;
    import java.util.concurrent.locks.LockSupport;
    
    public class MyLock {
        /**
         * 基于CAS(Compare And Swap)+LockSupport+AQS(Abstract Queue Synchronizer)写自己的锁
         */
        //获取锁
        private final AtomicInteger lockState = new AtomicInteger(0);
    
        //获取到锁的线程
        private Thread getLockThread = null;
    
        //没有获取到锁的线程集合
        private final ConcurrentLinkedDeque<Thread> concurrentLinkedDeque = new ConcurrentLinkedDeque<>();
        public void lock(){
            acquire();
        }
    
        public boolean acquire(){
            while (true) {
                //此行代码表示获取锁成功
                System.out.println(Thread.currentThread().getName() + " " + "CAS操作");
                if (compareAndSet(0, 1)) {
                    getLockThread = Thread.currentThread();
                    return true;
                }
                //获取锁失败
                Thread thread = Thread.currentThread();
                concurrentLinkedDeque.add(thread);
                LockSupport.park();
            }
        }
    
        //再封装一遍Compare And Set
        private boolean compareAndSet(int expect,int update){
            return lockState.compareAndSet(expect,update);
        }
    
        //释放锁
        public boolean unlock(){
            if(getLockThread == null){
                return false;
            }
            if (getLockThread == Thread.currentThread()){
                boolean result =  compareAndSet(1,0);
                if(result){
                    //公平锁唤醒
                    Thread thread = concurrentLinkedDeque.getFirst();
                    System.out.println(thread.getName()+"被唤醒");
                    LockSupport.unpark(thread);
                    return true;
                }
            }
            return false;
        }
        public AtomicInteger getState(){
            return lockState;
        }
    }

创建Test01进行测试：

package JUC;
    
    public class Test01 {
        public static void main(String[] args) throws InterruptedException {
            MyLock myLock = new MyLock();
            myLock.lock();
            new Thread(() -> {
                System.out.println(Thread.currentThread().getName()+" "+"start");
                myLock.lock();
                System.out.println(Thread.currentThread().getName()+" "+"end");
            }).start();
            Thread.sleep(2000);
            myLock.unlock();
            System.out.println(myLock.getState().get());
        }
    }

运行结果：

![d9a23d0a0caa414399cc98a34325d3e1.png][]

--------------------

## 四、Semaphore(信号量)及其API ##

> 信号量（Semaphore）是一种用于控制多个并发进程或线程访问共享资源的同步机制。它主要用于解决并发编程中的互斥和同步问题。
> 
> Semaphore包含两个主要操作：P（wait）和V（signal）。
> 
> P（wait）操作会锁定信号量并减少其计数器的值。如果计数器的值变为负数，表示资源正在被其他进程或线程占用，当前进程或线程需要等待。P操作会阻塞进程或线程，直到可以获得资源并继续执行。
> 
> V（signal）操作会释放信号量并增加其计数器的值。如果有进程或线程正在等待该资源，V操作会唤醒其中一个，并让其继续执行。
> 
> 信号量中的计数器可以是任意整数值，通常用于表示可用的资源数量。当信号量为1时，也可以称为二进制信号量（binary semaphore），用于实现互斥锁（mutual exclusion）。
> 
> 使用信号量可以有效地控制对共享资源的访问，防止竞态条件（race condition）和资源冲突（resource conflict）的发生。例如，多个进程需要访问一个共享的数据结构时，可以使用信号量来确保同时只有一个进程可以访问该数据结构。
> 
> 需要注意的是，信号量机制是一种底层的同步原语，需要程序员手动管理它们的使用。使用不当可能导致死锁（deadlock）或其他并发问题，因此需要谨慎使用。

--------------------

## 五、CountDownLatch及其API ##

> **CountDownLatch**（倒计时门闩）是Java多线程编程中的一种同步辅助类，它用于控制一个或多个线程等待其他线程完成操作后再继续执行。
> 
> CountDownLatch的原理很简单，它的核心是一个倒计数器。当一个或多个线程调用CountDownLatch的await()方法时，它们会被阻塞，直到在CountDownLatch上调用countDown()方法的次数达到预设值，才会继续执行。
> 
> 使用CountDownLatch可以实现一些并发场景，比如等待多个线程完成某项任务后再开始下一步操作，或者等待多个线程都准备就绪后再同时执行。下面是CountDownLatch的基本用法：
> 
> 1. 创建一个CountDownLatch对象，并设置计数值，表示有多少个线程需要等待：
> 
>     CountDownLatch latch = new CountDownLatch(3);
>     ```
> 
> 2. 多个线程执行任务，任务执行完成后调用CountDownLatch的countDown()方法：
> 
>     latch.countDown();
>     ```
> 
> 3. 其他线程调用CountDownLatch的await()方法等待计数值变为0：
> 
>     latch.await();
>     ```
> 
> 4. 当计数值变为0时，等待的线程会被唤醒，继续执行后续操作。
> 
> CountDownLatch非常适用于一些需要等待多个线程完成任务的场景，比如主线程需要等待多个子线程完成初始化工作后再启动。它提供了一种简单而有效的线程同步方法，可以避免手动编写复杂的wait-notify代码。
> 
> 需要注意的是，CountDownLatch的计数值只能在初始化时设置一次，无法重置。一旦计数值减为0，再调用countDown()方法也不会有任何作用。如果需要可重复使用的计数器，可以考虑使用CyclicBarrier或Semaphore。

[d9a23d0a0caa414399cc98a34325d3e1.png]: https://img-blog.csdnimg.cn/d9a23d0a0caa414399cc98a34325d3e1.png