HOOOS

Java 多线程协作利器：CountDownLatch 深度解析与实战演练

0 87 老码农爱学习 Java并发编程CountDownLatch
Apple

你好，我是老码农，很高兴能和你一起探讨 Java 并发编程中的 CountDownLatch。在多线程的世界里，协调各个线程的运行至关重要。今天，我们就来深入了解一下 CountDownLatch 这个强大的工具，看看它如何帮助我们解决实际问题。

1. CountDownLatch 是什么？

CountDownLatch，直译过来就是“计数门闩”。它位于 java.util.concurrent 包下，是 Java 并发编程中一个非常实用的同步辅助工具。它的核心功能就像一个倒计时计数器，允许一个或多个线程等待其他线程完成操作。

基本原理：

  • 初始化： CountDownLatch 在创建时需要指定一个计数器的初始值。这个值代表需要完成的任务数量或者需要等待的线程数量。
  • countDown() 方法： 每当一个线程完成一个任务，或者准备就绪，就调用 countDown() 方法，将计数器的值减 1。
  • await() 方法： 一个或多个线程调用 await() 方法，开始等待。这些线程会被阻塞，直到计数器的值变为 0。
  • 计数器归零： 当计数器的值变为 0 时，所有等待的线程会被唤醒，继续执行。

简单来说： CountDownLatch 就像一个闸门，初始是关闭的。当闸门上的计数器倒计时到 0 时，闸门就会打开，允许所有等待的线程通过。

2. CountDownLatch 的核心方法

CountDownLatch 提供了几个核心方法，理解这些方法是使用 CountDownLatch 的关键。

  • CountDownLatch(int count)： 构造方法。创建一个 CountDownLatch 实例，并初始化计数器的值。 count 参数表示需要等待的线程或任务的数量。

    CountDownLatch latch = new CountDownLatch(3); // 创建一个计数器，初始值为 3

  • void await()： 使当前线程阻塞，直到计数器的值为 0。如果计数器的值为 0，则该方法立即返回。 await() 方法有多个重载版本，可以设置超时时间。

    latch.await(); // 等待计数器变为 0

    latch.await(10, TimeUnit.SECONDS); // 等待 10 秒，如果超时则继续执行

  • void countDown()： 将计数器的值减 1。如果计数器的值减为 0，则唤醒所有在 await() 方法上等待的线程。

    latch.countDown(); // 计数器减 1

  • long getCount()： 获取当前计数器的值。

    long count = latch.getCount(); // 获取当前计数器的值

3. CountDownLatch 的典型应用场景

CountDownLatch 在许多多线程协作的场景中都非常有用。下面，我们结合具体的例子，来看看它的实际应用。

3.1 任务分解与合并

场景： 假设我们需要处理一批数据，可以将任务分解成多个子任务，每个子任务由一个线程处理。所有子任务完成后，再将结果合并。

代码示例：

import java.util.concurrent.CountDownLatch; 
import java.util.concurrent.ExecutorService; 
import java.util.concurrent.Executors; 

public class TaskDecomposition {

    public static void main(String[] args) throws InterruptedException {
        int taskCount = 3; // 子任务数量
        CountDownLatch latch = new CountDownLatch(taskCount);
        ExecutorService executor = Executors.newFixedThreadPool(taskCount);

        for (int i = 1; i <= taskCount; i++) {
            final int taskId = i;
            executor.submit(() -> {
                try {
                    System.out.println("Task " + taskId + " started.");
                    Thread.sleep((long) (Math.random() * 2000)); // 模拟任务处理时间
                    System.out.println("Task " + taskId + " finished.");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    latch.countDown(); // 任务完成，计数器减 1
                }
            });
        }

        System.out.println("Waiting for all tasks to complete...");
        latch.await(); // 主线程等待所有子任务完成
        System.out.println("All tasks completed. Result merging...");

        executor.shutdown();
    }
}

运行结果：

Waiting for all tasks to complete...
Task 1 started.
Task 2 started.
Task 3 started.
Task 3 finished.
Task 2 finished.
Task 1 finished.
All tasks completed. Result merging...

分析：

  • 我们创建了一个 CountDownLatch，初始值为 3，对应 3 个子任务。
  • 每个子任务执行完毕后，都会调用 latch.countDown()。
  • 主线程调用 latch.await()，等待所有子任务完成。只有当 count 变为 0 时，主线程才会继续执行。

3.2 资源初始化

场景： 在系统启动时，需要初始化多个资源，例如数据库连接、缓存等等。这些资源的初始化需要并发执行，确保所有资源都准备好之后，才能启动主程序。

代码示例：

import java.util.concurrent.CountDownLatch; 
import java.util.concurrent.ExecutorService; 
import java.util.concurrent.Executors; 

public class ResourceInitialization {

    public static void main(String[] args) throws InterruptedException {
        int resourceCount = 3; // 资源数量
        CountDownLatch latch = new CountDownLatch(resourceCount);
        ExecutorService executor = Executors.newFixedThreadPool(resourceCount);

        // 模拟数据库连接初始化
        executor.submit(() -> {
            try {
                System.out.println("Initializing database connection...");
                Thread.sleep(1000); // 模拟初始化时间
                System.out.println("Database connection initialized.");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                latch.countDown();
            }
        });

        // 模拟缓存初始化
        executor.submit(() -> {
            try {
                System.out.println("Initializing cache...");
                Thread.sleep(500); // 模拟初始化时间
                System.out.println("Cache initialized.");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                latch.countDown();
            }
        });

        // 模拟其他资源初始化
        executor.submit(() -> {
            try {
                System.out.println("Initializing other resources...");
                Thread.sleep(700); // 模拟初始化时间
                System.out.println("Other resources initialized.");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                latch.countDown();
            }
        });

        System.out.println("Waiting for resource initialization...");
        latch.await(); // 等待所有资源初始化完成
        System.out.println("All resources initialized. Starting the application...");

        executor.shutdown();
    }
}

运行结果：

Waiting for resource initialization...
Initializing database connection...
Initializing cache...
Initializing other resources...
Cache initialized.
Other resources initialized.
Database connection initialized.
All resources initialized. Starting the application...

分析：

  • 我们创建了一个 CountDownLatch，初始值为 3，对应 3 个资源初始化任务。
  • 每个资源初始化任务完成后，调用 latch.countDown()。
  • 主线程调用 latch.await()，等待所有资源初始化完成。只有当 count 变为 0 时，主程序才能启动。

3.3 服务启动协调

场景： 在分布式系统中，多个服务需要协同启动。例如，服务 A 依赖于服务 B 的启动。我们可以使用 CountDownLatch 来协调服务的启动顺序。

代码示例：

import java.util.concurrent.CountDownLatch; 
import java.util.concurrent.ExecutorService; 
import java.util.concurrent.Executors; 

public class ServiceStartup {

    public static void main(String[] args) throws InterruptedException {
        CountDownLatch serviceBLatch = new CountDownLatch(1); // 服务 B 启动的门闩
        ExecutorService executor = Executors.newFixedThreadPool(2);

        // 启动服务 B
        executor.submit(() -> {
            try {
                System.out.println("Service B starting...");
                Thread.sleep(1500); // 模拟服务 B 启动时间
                System.out.println("Service B started.");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                serviceBLatch.countDown(); // 服务 B 启动完成
            }
        });

        // 启动服务 A，依赖于服务 B
        executor.submit(() -> {
            try {
                System.out.println("Service A waiting for Service B...");
                serviceBLatch.await(); // 等待服务 B 启动
                System.out.println("Service A starting...");
                Thread.sleep(1000); // 模拟服务 A 启动时间
                System.out.println("Service A started.");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        Thread.sleep(2000); // 确保服务 B 启动完成
        executor.shutdown();
    }
}

运行结果：

Service B starting...
Service A waiting for Service B...
Service B started.
Service A starting...
Service A started.

分析：

  • 服务 B 启动完成后，调用 serviceBLatch.countDown()。
  • 服务 A 调用 serviceBLatch.await()，等待服务 B 启动完成。只有当 count 变为 0 时，服务 A 才能继续启动。

4. CountDownLatch 的局限性

虽然 CountDownLatch 很好用，但它也有一些局限性，我们需要了解这些局限性，才能更好地选择合适的工具。

  • 一次性使用： CountDownLatch 的计数器只能使用一次。一旦计数器变为 0，就不能再重置。如果需要重复使用，需要重新创建一个新的 CountDownLatch。
  • 无法取消： await() 方法是阻塞的，而且无法被中断。如果线程在 await() 上阻塞了，只能等待计数器变为 0 或者超时。
  • 无法获取中间状态： CountDownLatch 只能判断任务是否完成，而不能获取任务的中间状态或者进度信息。
  • 缺乏灵活性： CountDownLatch 的功能相对简单，只能用于线程之间的同步。如果需要更复杂的线程协作，可能需要使用更强大的工具。

5. CountDownLatch 的替代方案

针对 CountDownLatch 的局限性，Java 提供了其他一些并发工具，可以用于解决更复杂的线程协作问题。下面介绍几种常用的替代方案。

5.1 CyclicBarrier

功能： CyclicBarrier 允许一组线程相互等待，直到所有线程都到达某个屏障点（barrier）。与 CountDownLatch 不同，CyclicBarrier 可以重复使用。

适用场景： 多个线程需要重复执行相同的任务，并在每次执行后进行同步。例如，计算一个矩阵的平均值，可以把矩阵分成多块，每个线程计算一块的平均值，然后所有线程同步，合并结果。

代码示例：

import java.util.concurrent.BrokenBarrierException; 
import java.util.concurrent.CyclicBarrier; 
import java.util.concurrent.ExecutorService; 
import java.util.concurrent.Executors; 

public class CyclicBarrierExample {

    public static void main(String[] args) {
        int threadCount = 3; // 线程数量
        CyclicBarrier barrier = new CyclicBarrier(threadCount, () -> {
            System.out.println("All threads reached the barrier. Performing final operation.");
        });
        ExecutorService executor = Executors.newFixedThreadPool(threadCount);

        for (int i = 0; i < threadCount; i++) {
            final int threadId = i;
            executor.submit(() -> {
                try {
                    System.out.println("Thread " + threadId + " is working...");
                    Thread.sleep((long) (Math.random() * 1000)); // 模拟工作时间
                    System.out.println("Thread " + threadId + " reached the barrier.");
                    barrier.await(); // 等待其他线程到达屏障
                    System.out.println("Thread " + threadId + " continues.");
                } catch (InterruptedException | BrokenBarrierException e) {
                    e.printStackTrace();
                }
            });
        }
        executor.shutdown();
    }
}

5.2 Semaphore

功能： Semaphore（信号量）用于控制对共享资源的访问。它可以限制同时访问某个资源的线程数量。

适用场景： 限制并发访问资源的线程数量，例如，限制数据库连接池的大小，或者限制同时下载文件的数量。

代码示例：

import java.util.concurrent.ExecutorService; 
import java.util.concurrent.Executors; 
import java.util.concurrent.Semaphore; 

public class SemaphoreExample {

    public static void main(String[] args) {
        int threadCount = 5; // 线程数量
        int permitCount = 2; // 许可数量
        Semaphore semaphore = new Semaphore(permitCount);
        ExecutorService executor = Executors.newFixedThreadPool(threadCount);

        for (int i = 0; i < threadCount; i++) {
            final int threadId = i;
            executor.submit(() -> {
                try {
                    System.out.println("Thread " + threadId + " is trying to acquire a permit.");
                    semaphore.acquire(); // 获取许可
                    System.out.println("Thread " + threadId + " acquired a permit.");
                    Thread.sleep(2000); // 模拟占用资源的时间
                    System.out.println("Thread " + threadId + " is releasing the permit.");
                    semaphore.release(); // 释放许可
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        executor.shutdown();
    }
}

5.3 Future 和 CompletionService

功能： Future 用于获取异步计算的结果。CompletionService 结合了 Executor 和 BlockingQueue 的功能，可以更方便地获取异步任务的结果。

适用场景： 提交异步任务，获取任务的执行结果，并处理任务的完成顺序。例如，并发执行多个 Web 请求，并按照请求的完成顺序处理结果。

代码示例（Future）：

import java.util.concurrent.Callable; 
import java.util.concurrent.ExecutorService; 
import java.util.concurrent.Executors; 
import java.util.concurrent.Future; 

public class FutureExample {

    public static void main(String[] args) throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(2);

        // 提交一个异步任务
        Future<Integer> future = executor.submit(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                System.out.println("Task is running...");
                Thread.sleep(2000); // 模拟任务执行时间
                return 100; // 返回结果
            }
        });

        // 主线程可以做其他事情
        System.out.println("Main thread is doing something else...");

        // 获取任务结果，如果任务尚未完成，则阻塞
        Integer result = future.get();
        System.out.println("Task result: " + result);

        executor.shutdown();
    }
}

代码示例（CompletionService）：

import java.util.concurrent.Callable; 
import java.util.concurrent.CompletionService; 
import java.util.concurrent.ExecutorCompletionService; 
import java.util.concurrent.ExecutorService; 
import java.util.concurrent.Executors; 
import java.util.concurrent.Future; 

public class CompletionServiceExample {

    public static void main(String[] args) throws Exception {
        int taskCount = 3; // 任务数量
        ExecutorService executor = Executors.newFixedThreadPool(taskCount);
        CompletionService<Integer> completionService = new ExecutorCompletionService<>(executor);

        // 提交多个任务
        for (int i = 0; i < taskCount; i++) {
            final int taskId = i;
            completionService.submit(new Callable<Integer>() {
                @Override
                public Integer call() throws Exception {
                    System.out.println("Task " + taskId + " is running...");
                    Thread.sleep((long) (Math.random() * 2000)); // 模拟任务执行时间
                    return taskId * 10; // 返回结果
                }
            });
        }

        // 按照任务完成的顺序获取结果
        for (int i = 0; i < taskCount; i++) {
            Future<Integer> future = completionService.take(); // 获取已完成的任务，阻塞直到有任务完成
            Integer result = future.get();
            System.out.println("Task result: " + result);
        }

        executor.shutdown();
    }
}

5.4 其他并发工具

除了上面介绍的，Java 还提供了许多其他的并发工具，例如：

  • Phaser： 用于控制多个阶段的并发任务。
  • StampedLock： 提供了乐观读锁和悲观锁，更灵活的读写锁机制。
  • BlockingQueue： 用于实现生产者-消费者模型。

选择合适的并发工具，需要根据具体的应用场景和需求进行评估。理解每种工具的特性和优缺点，才能更好地解决并发编程问题。

6. CountDownLatch 的使用技巧与注意事项

在使用 CountDownLatch 时，有一些技巧和注意事项可以帮助我们更好地利用它。

  • 初始化计数器的值： 计数器的值应该根据需要等待的线程或任务数量来确定。确保计数器的值是正确的，避免出现死锁或者程序无法正常结束的情况。
  • countDown() 的调用位置： 确保在每个线程或任务完成时，正确地调用 countDown() 方法。如果 countDown() 的调用位置不正确，可能会导致计数器值错误，从而影响程序的正确性。
  • await() 的超时时间： await() 方法可以设置超时时间。在某些场景下，设置超时时间可以避免线程无限期地阻塞。如果超时，程序可以采取相应的处理措施，例如重试或者报告错误。
  • 避免死锁： 在使用 CountDownLatch 时，要注意避免死锁。例如，如果多个线程之间存在循环依赖关系，并且使用了 CountDownLatch 进行同步，就可能导致死锁。
  • 合理使用线程池： 在使用 CountDownLatch 的时候，经常会结合线程池使用。合理地配置线程池的大小，可以提高程序的并发性能。同时，也要注意线程池的生命周期管理，避免资源泄露。
  • 异常处理： 在线程中使用 CountDownLatch 的时候，要注意异常处理。如果线程执行过程中发生了异常，需要确保在 finally 块中调用 countDown() 方法，保证计数器能够正确地减少。
  • 监控和调试： 在多线程程序中，监控和调试非常重要。可以使用一些工具来监控 CountDownLatch 的状态，例如计数器的值，以及等待的线程数量。这可以帮助我们快速地定位和解决问题。

7. 总结

CountDownLatch 是一个非常实用的多线程协作工具，它可以帮助我们协调多个线程的执行顺序，解决许多并发编程问题。通过本文的介绍，相信你对 CountDownLatch 已经有了深入的理解。我们学习了它的基本原理、核心方法、应用场景，以及局限性和替代方案。同时，我们也探讨了使用 CountDownLatch 的一些技巧和注意事项。

在实际开发中，我们需要根据具体的应用场景选择合适的并发工具。理解每种工具的特性和优缺点，才能写出高效、可靠的多线程程序。希望这篇文章能帮助你更好地掌握 CountDownLatch，并在并发编程的道路上越走越远！加油！

如果你有任何问题，或者想分享你的经验，欢迎在评论区留言！让我们一起学习，共同进步！

点评评价

captcha
健康