Java - AQS-CountDownLatch实现类（二）

前言

在Java中，AbstractQueuedSynchronizer（简称AQS）是一个用于实现同步器的抽象类，它为实现各种类型的同步器（如锁、信号量等）提供了基本的框架。AQS通过一个双向队列（等待队列）和一个整数变量（状态）来管理线程的排队和状态控制。

AQS的具体实现类主要有以下几种：

1. ReentrantLock：可重入锁的实现类，支持独占模式。
2. ReentrantReadWriteLock：可重入读写锁的实现类，支持共享和独占模式。
3. Semaphore：信号量的实现类，用于控制同时访问某个资源的线程数量。
4. CountDownLatch：倒计时门闩的实现类，用于等待一组线程完成某个操作后再执行。
5. CyclicBarrier：循环屏障的实现类，用于等待一组线程都到达某个状态点后再一起继续执行。
6. Phaser：分阶段屏障的实现类，用于协调多个线程在不同阶段的同步。
7. LockSupport：用于创建锁和其他同步类的基本线程阻塞原语。

其中，ReentrantLock和ReentrantReadWriteLock是比较常用的同步器实现类，用于提供可重入的独占锁和读写锁。Semaphore、CountDownLatch、CyclicBarrier和Phaser等同步器类则在特定场景下发挥重要作用，帮助线程协调、控制和同步操作。

这些AQS实现类可以根据具体的需求选择使用，通过继承AbstractQueuedSynchronizer类并实现其中的抽象方法，可以定制和扩展自定义的同步器。

CountDownLatch

简介

CountDownLatch是Java中的一个同步辅助类，用于实现线程间的等待和通知机制。它通过一个计数器来实现，该计数器被初始化为一个正整数，并且只能递减。线程在等待阶段通过调用await()方法等待计数器变为0，而其他线程在完成自己的任务后通过调用countDown()方法来减少计数器的值。当计数器的值变为0时，所有等待的线程都将被唤醒继续执行。
CountDownLatch的主要方法包括：

void await()：当前线程等待计数器的值变为0。如果计数器的值大于0，await()方法将导致线程阻塞，直到计数器的值变为0或被中断。
void countDown()：将计数器的值减1。每个调用countDown()方法的线程都会使计数器减少1。
long getCount()：获取当前计数器的值。
CountDownLatch通常用于以下场景：

启动多个线程等待某个任务的完成：主线程在创建需要等待的线程后，通过CountDownLatch的构造函数将计数器的值设置为等待的线程数。每个线程在完成任务后调用countDown()方法，计数器的值减1。主线程在需要等待的位置调用await()方法，直到计数器的值变为0，才会继续执行。
并行任务的等待和合并：多个线程同时执行某个任务，但是要求它们在继续执行前等待其他所有线程都完成。每个线程在完成自己的任务后调用countDown()方法，主线程调用await()方法等待所有线程完成。
测试并发性能：可以使用CountDownLatch来同时启动多个线程，然后在主线程中等待所有线程完成，以测试并发操作的性能。
通过使用CountDownLatch，我们可以实现线程之间的同步和协调，确保某些线程在其他线程完成任务后再继续执行，以及在需要等待多个线程完成后再进行下一步操作。

其中内部有一个类Sync可以看到该类继承AbstractQueuedSynchronizer，拥有AQS特性。
从构造函数来看Sync(int count)，构造时传递一个线程状态存储。

    //CountDownLatch的同步控制。使用AQS状态表示计数。private static final class Sync extends AbstractQueuedSynchronizer {private static final long serialVersionUID = 4982264981922014374L;Sync(int count) {setState(count);}int getCount() {return getState();}protected int tryAcquireShared(int acquires) {return (getState() == 0) ? 1 : -1;}protected boolean tryReleaseShared(int releases) {// Decrement count; signal when transition to zerofor (;;) {int c = getState();if (c == 0)return false;int nextc = c-1;if (compareAndSetState(c, nextc))return nextc == 0;}}}

函数

await

• await()函数是CountDownLatch的核心函数之一，用于使当前线程在计数器倒计数至零之前一直等待。如果线程被中断，则会抛出InterruptedException。

• 函数内部通过调用sync.acquireSharedInterruptibly(1)将操作转发给了Sync对象的acquireSharedInterruptibly方法。

public void await() throws InterruptedException {// 转发到sync对象上sync.acquireSharedInterruptibly(1);
}

acquireSharedInterruptibly

acquireSharedInterruptibly(int arg)方法在AQS中定义，用于在获取共享资源时可中断地等待。如果线程被中断，则会抛出InterruptedException。

acquireSharedInterruptibly(int arg)方法内部先检查线程的中断状态，如果被中断，则抛出InterruptedException。

public final void acquireSharedInterruptibly(int arg)throws InterruptedException {if (Thread.interrupted())throw new InterruptedException();if (tryAcquireShared(arg) < 0)doAcquireSharedInterruptibly(arg);
}

tryAcquireShared

如果通过tryAcquireShared(arg)方法尝试获取共享资源失败（返回值小于0），则调用doAcquireSharedInterruptibly(arg)方法进行进一步的等待操作。
tryAcquireShared(int acquires)是Sync类的方法，它根据AQS的状态来判断是否可以获取共享资源。在CountDownLatch的实现中，该方法简单地判断AQS的状态是否为0，如果为0，则返回1，表示可以获取共享资源；否则返回-1，表示不能获取共享资源

protected int tryAcquireShared(int acquires) {return (getState() == 0) ? 1 : -1;
}

示例1

创建了一个CountDownLatch对象，并将其初始计数值设置为3（numOfThreads）。然后，我们创建了3个工作线程（Worker类的实例），每个工作线程模拟执行任务的时间。每个工作线程完成任务后，调用countDown()方法来减少计数器的值。
**主线程通过调用latch.await()来等待所有工作线程完成。**当计数器的值变为0时，主线程被唤醒，并打印出"All workers have completed their tasks."的消息。
该示例演示了如何使用CountDownLatch实现主线程等待多个工作线程完成任务后再继续执行的场景。

import java.util.concurrent.CountDownLatch;public class CountDownLatchExample {public static void main(String[] args) {int numOfThreads = 3;CountDownLatch latch = new CountDownLatch(numOfThreads);for (int i = 0; i < numOfThreads; i++) {Thread thread = new Thread(new Worker(latch));thread.start();}try {latch.await(); // 主线程等待所有工作线程完成System.out.println("All workers have completed their tasks.");} catch (InterruptedException e) {e.printStackTrace();}}static class Worker implements Runnable {private final CountDownLatch latch;public Worker(CountDownLatch latch) {this.latch = latch;}@Overridepublic void run() {// 模拟每个工作线程执行任务的时间try {Thread.sleep(2000);System.out.println("Worker completed its task.");} catch (InterruptedException e) {e.printStackTrace();} finally {latch.countDown(); // 每个工作线程完成任务后调用countDown()}}}
}

}
在上面的示例中，我们创建了一个CountDownLatch对象，并将其初始计数值设置为3（numOfThreads）。然后，我们创建了3个工作线程（Worker类的实例），每个工作线程模拟执行任务的时间。每个工作线程完成任务后，调用countDown()方法来减少计数器的值。
主线程通过调用latch.await()来等待所有工作线程完成。当计数器的值变为0时，主线程被唤醒，并打印出"All workers have completed their tasks."的消息。
该示例演示了如何使用CountDownLatch实现主线程等待多个工作线程完成任务后再继续执行的场景。

示例2

业务场景：某个业务操作非常耗时，但又必须等这个操作结束后才能进行后续操作。

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.Stream;/*** 多线程任务处理工具类*/
public class TaskDisposeUtils {//并行线程数public static final int POOL_SIZE;static {//判断核心线程数 如果机器的核心线程数大于5则用机器核心线程数POOL_SIZE = Integer.max(Runtime.getRuntime().availableProcessors(), 5);}/*** 并行处理，并等待结束**/public static <T> void dispose(List<T> taskList, Consumer<T> consumer) throws InterruptedException {dispose(true, POOL_SIZE, taskList, consumer);}/*** 并行处理，并等待结束**/public static <T> void dispose(boolean moreThread, int poolSize, List<T> taskList, Consumer<T> consumer) throws InterruptedException {if (CollectionUtils.isEmpty(taskList)) {return;}//如果是多线程且核心线程数大于一则进入方法if (moreThread && poolSize > 1) {poolSize = Math.min(poolSize, taskList.size());ExecutorService executorService = null;try {//新建一个固定大小的线程池 核心线程数为poolSizeexecutorService = Executors.newFixedThreadPool(poolSize);//juc工具类 用于让必须所有任务都处理完后才进行下一步CountDownLatch countDownLatch = new CountDownLatch(taskList.size());for (T item : taskList) {executorService.execute(() -> {try {//消费任务consumer.accept(item);} finally {//处理完后减一countDownLatch.countDown();}});}//在此等待 当countDownLatch变成0后才继续进行下一步countDownLatch.await();} finally {if (executorService != null) {executorService.shutdown();}}} else {for (T item : taskList) {consumer.accept(item);}}}public static void main(String[] args) throws InterruptedException {//生成1-10的10个数字，放在list中，相当于10个任务List<Integer> list = Stream.iterate(1, a -> a + 1).limit(10).collect(Collectors.toList());JSONObject object=new JSONObject();object.put("name","sss");//启动多线程处理list中的数据，每个任务休眠时间为list中的数值
//        Consumer<Integer> c= item -> {
//            try {
//                long startTime = System.currentTimeMillis();
//                object.put("s",item);
//                TimeUnit.SECONDS.sleep(item);
//                long endTime = System.currentTimeMillis();
//                System.out.println(object.toJSONString());
//                System.out.println(System.currentTimeMillis() + ",任务" + item + "执行完毕，耗时:" + (endTime - startTime));
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//        };TaskDisposeUtils.dispose(list, item -> {try {long startTime = System.currentTimeMillis();object.put("s",item);TimeUnit.SECONDS.sleep(item);long endTime = System.currentTimeMillis();System.out.println(object.toJSONString());System.out.println(System.currentTimeMillis() + ",任务" + item + "执行完毕，耗时:" + (endTime - startTime));} catch (InterruptedException e) {e.printStackTrace();}});//上面所有任务处理完毕完毕之后，程序才能继续System.out.println(list + "中的任务都处理完毕!");}
}