JUC并发编程(八)——AQS(并发灵魂)

AQS介绍

在 Java 并发编程中，AbstractQueuedSynchronizer（AQS）是一个非常重要的组件。AQS 是 JDK 提供的一个框架，用于实现基于 FIFO（First In, First Out）等待队列的阻塞锁和同步器，例如 ReentrantLock、Semaphore、CountDownLatch 等。本文将从多个角度深入解析 AQS 的工作原理及其在并发编程中的应用。

AQS 是一个抽象类，它定义了一套多线程访问共享资源的同步机制。其核心思想是：如果被请求的共享资源空闲，则将当前请求资源的线程设置为有效的工作线程，并且将共享资源设置为锁定状态；如果被请求的共享资源被占用，则将当前线程加入到等待队列中，并挂起直到被唤醒。

AQS核心功能

AQS 是一个用于构建锁和同步器的基础框架，它通过一个先进先出的等待队列管理多个线程的同步操作。AQS 提供了以下核心功能：

资源状态管理：通过一个整数表示同步状态，子类可以定义该状态的意义和操作方式。
线程排队管理：通过 FIFO 等待队列管理多个线程的竞争和等待。
提供模板方法：AQS 提供了一系列模板方法，子类可以通过实现这些方法来定义具体的同步机制。

AQS 设计理念

AQS 的设计理念是将同步状态和线程队列的管理逻辑抽象出来，以便不同类型的同步器可以重用这些逻辑。这样，开发者可以集中精力于具体的同步器逻辑实现，而无需关心底层的线程排队和状态管理细节。

AQS 的工作原理

同步状态

AQS 使用一个 volatile 类型（多线程可见）的整数变量（state）来表示同步状态。这个状态可以由子类定义其具体意义，例如：

对于独占锁，state 表示锁的持有状态（0 表示未持有，1 表示持有）。
对于共享锁，state 表示当前持有的读锁数量。
AQS 提供了一系列的方法来操作同步状态，如 getState()、setState(int newState) 和 compareAndSetState(int expect, int update)。

等待队列

AQS 使用一个 FIFO 等待队列来管理处于等待状态的线程。当线程请求无法获取同步状态时，会被加入等待队列。等待队列由一个双向链表实现，每个节点（Node）表示一个等待线程。

模板方法

AQS 提供了以下几个模板方法，子类可以通过覆盖这些方法来实现具体的同步器逻辑：

tryAcquire(int arg)：尝试获取同步状态。对于独占模式，如果成功获取返回 true，否则返回 false。
tryRelease(int arg)：尝试释放同步状态。对于独占模式，如果成功释放返回 true，否则返回 false。
tryAcquireShared(int arg)：尝试共享模式下获取同步状态。如果成功获取返回一个大于等于 0 的值，否则返回一个小于 0 的值。
tryReleaseShared(int arg)：尝试共享模式下释放同步状态。如果成功释放返回 true，否则返回 false。

AQS 的应用

AbstractQueuedSynchronizer工具包的实现方法：
JUC并发编程(八)——AQS(并发灵魂)-Nickの博客

ReentrantLock

ReentrantLock 是一种基于 AQS 实现的可重入独占锁。它有两种模式：公平模式和非公平模式。

公平模式

公平模式下，ReentrantLock 按照请求顺序获取锁。具体实现是在 tryAcquire 方法中检查等待队列中的前驱节点，如果没有前驱节点或前驱节点的线程已经被唤醒，则尝试获取锁。

非公平模式

非公平模式下，ReentrantLock 不保证按照请求顺序获取锁，而是直接尝试获取锁。如果失败，再进入等待队列。

Semaphore

Semaphore 是一种基于 AQS 实现的计数信号量，用于控制对资源的访问数量。它可以在独占模式和共享模式下使用。

在独占模式下，Semaphore 允许一个线程访问资源。在共享模式下，Semaphore 允许多个线程同时访问资源，其核心逻辑通过 tryAcquireShared 和 tryReleaseShared 方法实现。

CountDownLatch

CountDownLatch 是一种基于 AQS 实现的同步工具，它允许一个或多个线程等待一组操作完成。CountDownLatch 通过一个计数器来实现这一功能，每当一个操作完成时，计数器减 1，当计数器减到 0 时，所有等待的线程被唤醒。

CountDownLatch 主要使用了 AQS 的共享模式，通过 tryAcquireShared 方法判断计数器是否为 0，如果是，则获取同步状态，否则进入等待队列。

AQS 的优势

高度可扩展性
AQS 提供了一套通用的同步状态管理和线程排队机制，开发者可以通过扩展 AQS 实现各种复杂的同步器。AQS 的设计使得它具有高度的可扩展性，可以适应不同的并发编程需求。
可靠性和性能
AQS 使用 CAS（Compare-And-Swap）操作来管理同步状态，保证了操作的原子性和线程安全。同时，AQS 的等待队列机制减少了线程竞争，提高了并发性能。
简化开发
通过 AQS，开发者可以专注于同步器的具体逻辑实现，而无需关心底层的线程排队和状态管理细节。这大大简化了同步器的开发过程，提高了开发效率。

AQS的用法

第一步：写一个类，想好协作的逻辑，实现获取/释放方法。
第二步：内部写一个Sync类继承AbstractQueuedSynchronizer
第三步：根据是否独占来重写tryAcquire/tryRelease或tryAcquireShared(int acquires)和tryReleaseShared(intreleases)等方法，在之前写的获取/释放方法中调用AQS的acquire/release或者Shared方法

AQS在CountDownLatch的应用

内部类Sync继承AQS
JUC并发编程(八)——AQS(并发灵魂)-Nickの博客
构造函数
getCount
countDown
await

调用CountDownLatch的await方法时，便会尝试获取"共享
锁“，不过一开始是获取不到该锁的，于是线程被阻塞。
而“共享锁”可获取到的条件，就是”锁计数器“的值为0。
而“锁计数器”的初始值为count，每当一个线程调用该CountDownLatch对象的countDown()方法时，才将“锁计数器”-1
count个线程调用countDown()之后，"锁计数器”才为0，
而前面提到的等待获取共享锁的线程才能继续运行。

AQS在Semaphore的应用

在Semaphore中，state表示许可证的剩余数量
看tryAcquire方法，判断nonfairTryAcquireShared大于等于
0的话，代表成功
这里会先检查剩余许可证数量够不够这次需要的，用减法来计算，如果直接不够，那就返回负数，表示失败，如果够了，就用自旋加compareAndSetState来改变state状态，直到改变成功就返回正数；或者是期间如果被其他人修改了导致剩余数量不够了，那也返回负数代表获取失败

JUC并发编程(八)——AQS(并发灵魂)-Nickの博客

AQS在ReentrantLock的应用

释放锁的方法tryRelease
由于是可重入的，所以state代表重入的次数，每次释放锁，先判断是不是当前持有锁的线程释放的，如果不是就抛异常，如果是的话，重入次数就减一，如果减到了0，就说明完全释放了，于是free就是true，并且把state设置为0。
JUC并发编程(八)——AQS(并发灵魂)-Nickの博客

加锁的方法tryAcquire
先判断state是否等于0，并且判断当前线程是否是持有锁的线程，如果都不是，代表目前拿不到这把锁，放在等待队列等待，合适时间进行唤醒
JUC并发编程(八)——AQS(并发灵魂)-Nickの博客

自己用AQS实现一个简单的线程协作器

利用AQS实现一个类似于CountDownLatch的门闩

package aqs;

import java.util.concurrent.locks.AbstractQueuedSynchronizer;

/**
 * 描述：     自己用AQS实现一个简单的线程协作器
 */
public class OneShotLatch {

    private final Sync sync = new Sync();

    public void signal() {
        sync.releaseShared(0);
    }
    public void await() {
        sync.acquireShared(0);
    }

    private class Sync extends AbstractQueuedSynchronizer {

        @Override
        protected int tryAcquireShared(int arg) {
            return (getState() == 1) ? 1 : -1;
        }

        @Override
        protected boolean tryReleaseShared(int arg) {
           setState(1);

           return true;
        }
    }

    public static void main(String[] args) throws InterruptedException {
        OneShotLatch oneShotLatch = new OneShotLatch();
        for (int i = 0; i < 10; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName()+"尝试获取latch，获取失败那就等待");
                    oneShotLatch.await();
                    System.out.println("开闸放行"+Thread.currentThread().getName()+"继续运行");
                }
            }).start();
        }
        Thread.sleep(5000);
        oneShotLatch.signal();

        new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println(Thread.currentThread().getName()+"尝试获取latch，获取失败那就等待");
                oneShotLatch.await();
                System.out.println("开闸放行"+Thread.currentThread().getName()+"继续运行");
            }
        }).start();
    }
}

运行结果：
JUC并发编程(八)——AQS(并发灵魂)-Nickの博客

使用AQS state表示闸门状态
await()通过acquireShared进入AQS队列等待
signal()通过releaseShared唤醒所有线程
使用共享模式允许多个线程同时通过
state一旦变为1不会再变回0
所以oneShotLatch作用是一次性的