Java多线程实战-CountDownLatch模拟压测实现

news2024/10/6 22:23:39

🏷️个人主页：牵着猫散步的鼠鼠

🏷️系列专栏：Java全栈-专栏

🏷️本系列源码仓库：多线程并发编程学习的多个代码片段(github)

🏷️个人学习笔记，若有缺误，欢迎评论区指正

前言

CountDownLatch 的基本原理

CountDownLatch 基本用法

CountDownLatch 示例

使用CountDownLatch实现压测

总结

前言

当多个线程需要协调和同步执行任务时，Java 中的 CountDownLatch（倒计时器）是一个常用的工具类。它可以帮助开发者实现线程之间的同步，确保某些线程在其他线程完成任务后再继续执行。本文将介绍 CountDownLatch 的基本原理、用法以及示例代码，最后会使用CountDownLatch完成一个简单的压测实现。

CountDownLatch 的基本原理

CountDownLatch 是基于计数器的原理实现的，它内部维护了一个整型的计数器。创建一个 CountDownLatch 对象时，需要指定一个初始计数值，该计数值表示需要等待的线程数量。每当一个线程完成了其任务，它调用 CountDownLatch 的 countDown() 方法，计数器的值就会减一。当计数器的值变成 0 时，等待的线程就会被唤醒，继续执行它们的任务。

CountDownLatch 基本用法

CountDownLatch 在多线程编程中有广泛的应用场景，例如主线程等待所有子线程完成任务后再继续执行，多个线程协同完成一个任务等。以下是 CountDownLatch 的基本用法：

创建 CountDownLatch 对象，并指定初始计数值。

CountDownLatch latch = new CountDownLatch(3); // 初始计数值为 3

创建需要等待的线程，线程完成任务后调用 countDown() 方法。

Runnable task = new Runnable() {
    @Override
    public void run() {
        // 线程任务逻辑
        // ...
        latch.countDown(); // 完成任务后调用 countDown()
    }
};

创建等待线程，等待计数器的值变成 0 后再继续执行。

try {
    latch.await(); // 等待计数器的值变成 0
    // 继续执行需要等待的线程后续逻辑
} catch (InterruptedException e) {
    // 处理中断异常
    e.printStackTrace();
}

CountDownLatch 示例

下面再通过一个简单的示例代码来演示 CountDownLatch 的用法。假设有一个需求，需要三个工人协同完成一项任务，主线程需要等待三个工人完成任务后才能继续执行。示例代码如下：

import java.util.concurrent.CountDownLatch;

public class CountDownLatchExample {
    public static void main(String[] args) {
        final int workerCount = 3;
        final CountDownLatch latch = new CountDownLatch(workerCount);

        // 创建并启动三个工人线程
        for (int i = 0; i < workerCount; i++) {
            Worker worker = new Worker(latch, "Worker " + (i + 1));
            new Thread(worker).start();
        }

        try {
            System.out.println("Main thread is waiting for workers to finish...");
            latch.await(); // 主线程等待三个工人线程完成任务
            System.out.println("All workers have finished, main thread continues...");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    static class Worker implements Runnable {
        private final CountDownLatch latch;
        private final String name;

        public Worker(CountDownLatch latch, String name) {
            this.latch = latch;
            this.name = name;
        }

        @Override
        public void run() {
            System.out.println(name + " starts working...");
            // 模拟工作耗时
            try {
                Thread.sleep((long) (Math.random() * 10000));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(name + " finishes working.");
            latch.countDown(); // 工人完成任务后调用 countDown()
        }
    }
}

在上述代码中，首先创建了一个 CountDownLatch 对象，并指定初始计数值为 3。然后创建了三个工人线程并启动，每个工人线程模拟完成一项工作后调用 countDown() 方法，计数器的值减一。最后，主线程调用 await() 方法等待计数器的值变成 0，表示所有工人都完成任务后再继续执行。

运行该程序后，输出结果如下：

Worker 1 starts working...  
Worker 2 starts working...  
Worker 3 starts working...  
Main thread is waiting for workers to finish...  
Worker 2 finishes working.  
Worker 1 finishes working.  
Worker 3 finishes working.  
All workers have finished, main thread continues...

可以看到，主线程在三个工人线程完成任务后才继续执行，并且所有工人线程的完成顺序是不确定的，但主线程会一直等待直到所有工人完成任务。

扩展：上面的worker线程运行是没有顺序的，我们可以使用join()来使线程有序等待上一个线程运行结束。

public static void main(String[] args) throws InterruptedException {
    final int workerCount = 3;
    final CountDownLatch latch = new CountDownLatch(workerCount);
    System.out.println("Main thread is waiting for workers to finish...");
    // 创建并启动三个工人线程
    for (int i = 0; i < workerCount; i++) {
        Worker worker = new Worker(latch, "Worker " + (i + 1));
        Thread t = new Thread(worker);
        t.start();
        t.join(); // 等待上一个线程执行完成
    }
    try {
        latch.await(); // 主线程等待三个工人线程完成任务
        System.out.println("All workers have finished, main thread continues...");
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

输出结果：

Main thread is waiting for workers to finish...
Worker 1 starts working...
Worker 1 finishes working.
Worker 2 starts working...
Worker 2 finishes working.
Worker 3 starts working...
Worker 3 finishes working.
All workers have finished, main thread continues...

CountDownLatch 和 join 的作用和使用方式不同。CountDownLatch 用于等待多个线程完成任务后再继续执行，而 join 用于等待一个线程执行完毕后再继续执行。另外，CountDownLatch 是基于计数器的实现，可以灵活地控制线程的数量和完成顺序；而 join 方法只能等待单个线程执行完毕。

使用CountDownLatch实现压测

我们以上讲解了CountDownLatch的简单使用，接下来就利用CountDownLatch实现一个简单的压测程序，小伙伴可以在此基础上进行扩展

@Slf4j
public class concurrenceTest {
    public static void concurrenceTest() {
        /**
         * 模拟高并发情况代码
         */
        final AtomicInteger atomicInteger = new AtomicInteger(0);
        final CountDownLatch countDownLatch = new CountDownLatch(1000); // 相当于计数器，当所有都准备好了，再一起执行，模仿多并发，保证并发量
        final CountDownLatch countDownLatch2 = new CountDownLatch(1000); // 保证所有线程执行完了再打印atomicInteger的值
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        try {
            for (int i = 0; i < 1000; i++) {
                executorService.submit(() -> {
                    try {
                        countDownLatch.await(); //一直阻塞当前线程，直到计时器的值为0,保证同时并发
                    } catch (InterruptedException e) {
                        log.error(e.getMessage(), e);
                    }
                    //每个线程增加1000次，每次加1
                    for (int j = 0; j < 1000; j++) {
                        // 用业务代码替换
                        atomicInteger.incrementAndGet();
                    }
                    countDownLatch2.countDown();
                });
                countDownLatch.countDown();
            }
            countDownLatch2.await();// 保证所有线程执行完
            executorService.shutdown();
            log.info("atomicInteger的值为:{}", atomicInteger.get());
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    public static void main(String[] args) {
        concurrenceTest();
    }
}

在上述代码中，我们循环添加1000个任务到executorService 线程池中，并让每一个线程在加入到线程池后阻塞等待，每加入一个线程countDownLatch就减一，直到全部线程都加入到线程池中，线程池中的线程会停止阻塞开始执行，也就模拟了并发场景，可以自行调整线程池的线程数大小来调节压测压力。