Java - AQS-CountDownLatch实现类（二）

迷南。 2023-10-10 23:58 21阅读 0赞

### 前言 ###

在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 zero
                for (;;) {
        
                    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;
            }
    
            @Override
            public 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 {
        
                    //新建一个固定大小的线程池 核心线程数为poolSize
                    executorService = 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 + "中的任务都处理完毕!");
        }
    }