提供了编程的基础技术教程

zl程序教程

您现在的位置是：首页 >  其他

当前栏目

AQS同步组件-CountDownLatch解析和案例

案例同步组件 解析 AQS CountDownLatch
2023-06-13 09:13:24 时间

CountDownLatch原理

CountDownLatch是通过一个计数器来实现的，计数器的初始化值为线程的数量。每当一个线程完成了自己的任务后，计数器的值就相应得减1。当计数器到达0时，表示所有的线程都已完成任务，然后在闭锁上等待的线程就可以恢复执行任务。CountDownLatch可以起到阻塞线程，并保证线程在满足某种特定的情况后继续执行。

CountDownLatch 的两种典型用法

  • 某一线程在开始运行前等待n个线程执行完毕。将 CountDownLatch 的计数器初始化为n ：new CountDownLatch(n)，每当一个任务线程执行完毕，就将计数器减1 countdownlatch.countDown()，当计数器的值变为0时，在CountDownLatch上 await()的线程就会被唤醒。一个典型应用场景就是启动一个服务时，主线程需要等待多个组件加载完毕，之后再继续执行。即可以通过初始化，定义线程个数。
  • 实现多个线程开始执行任务的最大并行性。注意是并行性，不是并发，强调的是多个线程在某一时刻同时开始执行。类似于赛跑，将多个线程放到起点，等待发令枪响，然后同时开跑。做法是初始化一个共享的 CountDownLatch 对象，将其计数器初始化为 1 ：new CountDownLatch(1)，多个线程在开始执行任务前首先 coundownlatch.await()，当主线程调用 countDown() 时，计数器变为0，多个线程同时被唤醒。

源码分析

/**
     * 构造器中的计数值(count)实际上就是闭锁需要等待的线程数量。这个值只能被设置一次，
     *而且CountDownLatch没有提供任何机制去重新设置这个计数值。
     *
     * @param count 在线程可以通过 {@link #await} 之前必须调用 {@link #countDown} 的次数
     * @throws IllegalArgumentException 如果给定参数小于0则抛出异常
     */
    public CountDownLatch(int count) {
        if (count < 0) throw new IllegalArgumentException("count < 0");
        this.sync = new Sync(count);
    }
/**
     * 使当前线程等待，直到计数器为零，除非当前线程被中断
     * 1.当前计数为零，则此方法立即返回
     * 2.如果当前计数大于零，则当前线程将因线程调度目的而被禁用并处于休眠状态，
     * 直到发生如下可能：
     *	 由于调用了 {@link #countDown} 方法，计数达到零
     *   线程等待被中断。
     * @throws 如果当前线程在等待时被中断
     */
    public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }
/**    
	 * 可以设置等待时间，超过这个时间就会执行，不会等到计数器变为0,但是
     * 之前给定的线程还是会执行完
     *
     * @param timeout 等待时间长度
     * @param unit 等待时间单位
     * @return 
     * @throws 如果当前线程在等待时被中断
     */
public boolean await(long timeout, TimeUnit unit)
    throws InterruptedException {
    return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}

与CountDownLatch的第一次交互是主线程等待其他线程。主线程必须在启动其他线程后立即调用CountDownLatch.await()方法。这样主线程的操作就会在这个方法上阻塞，直到其他线程完成各自的任务。 其他N 个线程必须引用闭锁对象，因为他们需要通知CountDownLatch对象，他们已经完成了各自的任务。这种通知机制是通过 CountDownLatch.countDown()方法来完成的;每调用一次这个方法，在构造函数中初始化的count值就减1。所以当N个线程都调用了这个方法，count的值等于0，然后主线程就能通过await()方法，恢复执行自己的任务。 注意：

  • CountDownLatch的构造函数 CountDownLatch countDownLatch = new CountDownLatch(7); //7表示需要等待执行完毕的线程数量。
  • 在每一个线程执行完毕之后，都需要执行 countDownLatch.countDown() 方法，不然计数器就不会准确;
  • 只有所有的线程执行完毕之后，才会执行 countDownLatch.await() 之后的 代码;
  • CountDownLatch 阻塞的是主线程;

CountDownLatch 的使用示例

@Slf4j
public class CountDownLatchExample1 {
    /**
     * 线程数量
     */
    private final static int threadCount = 200;

    public static void main(String[] args) throws Exception {

        ExecutorService exec = Executors.newCachedThreadPool();

        final CountDownLatch countDownLatch = new CountDownLatch(threadCount);

        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    test(threadNum);
                } catch (Exception e) {
                    log.error("exception", e);
                } finally {
                    // 表示一个请求已经完成
                    countDownLatch.countDown();
                }
            });
        }
        //使当前线程等待，直到计数器为零，除非当前线程被中断
        countDownLatch.await();
        //当这200个请求被处理完成之后，才会执行
        log.info("finish");
        exec.shutdown();
    }

    private static void test(int threadNum) throws Exception {
        // 模拟请求的耗时操作
        Thread.sleep(100);
        log.info("{}", threadNum);
        Thread.sleep(100);
    }
}

上面的代码中，我们定义了请求的数量为200，当这200个请求被处理完成之后，才会执行System.out.println(“finish”);。

//可以设置等待时间，超过这个时间就会执行，不会等到计数器变为0,但是之前给定的线程还是会执行完
countDownLatch.await(20, TimeUnit.MILLISECONDS);

上面代码中其他跟第一个代码一致，使用了wait设置等待一定时间后继续执行方法。

/**
 * CountDownLatch 模拟并发调用多个任务
 *
 * @author zjq
 */
@Slf4j
public class CountDownLatchExample3 {

    public static void main(String[] args) throws InterruptedException {

        CountDownLatch countDownLatch = new CountDownLatch(2) {
            @Override
            public void await() throws InterruptedException {
                super.await();
                log.info("其他线程执行完毕后主线程执行的内容");
                log.info("threadName:{},", Thread.currentThread().getName() + " count down is ok");
            }
        };

        Thread thread1 = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(1000);
                    log.info(Thread.currentThread().getName() + "任务已完成");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    //计数器减1
                    countDownLatch.countDown();
                }
            }
        }, "thread111");

        Thread thread2 = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(2000);
                    log.info(Thread.currentThread().getName() + "任务已完成");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    //计数器减1
                    countDownLatch.countDown();
                }
            }
        }, "thread222");


        thread1.start();
        thread2.start();

        countDownLatch.await();
        log.info("====everything is end====");
    }

}

上述代码执行输出结果：

[thread111] INFO com.zjq.concurrency.aqs.CountDownLatchExample3 - thread111任务已完成
[thread222] INFO com.zjq.concurrency.aqs.CountDownLatchExample3 - thread222任务已完成
[main] INFO com.zjq.concurrency.aqs.CountDownLatchExample3 - 其他线程执行完毕后主线程执行的内容
[main] INFO com.zjq.concurrency.aqs.CountDownLatchExample3 - threadName:main count down is ok,
[main] INFO com.zjq.concurrency.aqs.CountDownLatchExample3 - ====everything is end====
/**
 * CountDownLatch 模拟多个任务并发执行完毕后等待主线程发令同时执行后续操作
 *
 * @author zjq
 */
@Slf4j
public class CountDownLatchExample4 {

    public static void main(String[] args) throws InterruptedException {

        CountDownLatch countDownLatch = new CountDownLatch(1);
        for (int i = 0; i < 5; i++) {
            new Thread(() -> {
                try {
                    //所有线程阻塞在这，等待主线程号令
                    log.info(Thread.currentThread().getName() + "已准备完毕！！");
                    countDownLatch.await();
                    log.info("【" + Thread.currentThread().getName() + "】" + "开始执行……");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }
        // 主线程准备发令
        Thread.sleep(2000);
        log.info(Thread.currentThread().getName() + "发号施令，给我冲！！");
        // 主线程：执行发令
        countDownLatch.countDown();
    }

}

实现最大并行性输出结果：

[Thread-3] INFO com.zjq.aqs.CountDownLatchExample4 - Thread-3已准备完毕！！
[Thread-0] INFO com.zjq.aqs.CountDownLatchExample4 - Thread-0已准备完毕！！
[Thread-2] INFO com.zjq.aqs.CountDownLatchExample4 - Thread-2已准备完毕！！
[Thread-1] INFO com.zjq.aqs.CountDownLatchExample4 - Thread-1已准备完毕！！
[Thread-4] INFO com.zjq.aqs.CountDownLatchExample4 - Thread-4已准备完毕！！
[main] INFO com.zjq.aqs.CountDownLatchExample4 - main发号施令，给我冲！！
[Thread-3] INFO com.zjq.aqs.CountDownLatchExample4 - 【Thread-3】开始执行……
[Thread-1] INFO com.zjq.aqs.CountDownLatchExample4 - 【Thread-1】开始执行……
[Thread-0] INFO com.zjq.aqs.CountDownLatchExample4 - 【Thread-0】开始执行……
[Thread-2] INFO com.zjq.aqs.CountDownLatchExample4 - 【Thread-2】开始执行……
[Thread-4] INFO com.zjq.aqs.CountDownLatchExample4 - 【Thread-4】开始执行……

CountDownLatch 的不足

CountDownLatch是一次性的，计数器的值只能在构造方法中初始化一次，之后没有任何机制再次对其设置值，当CountDownLatch使用完毕后，它不能再次被使用。

CountDownLatch应用场景

(1)实现最大的并行性:有时我们想同时启动多个线程，实现最大程度的并行性。例 如，我们想测试一个单例类。如果我们创建一个初始计数为1的CountDownLatch，并 让所有线程都在这个锁上等待，那么我们可以很轻松地完成测试。我们只需调用 一次 countDown()方法就可以让所有的等待线程同时恢复执行。 (2)开始执行前等待n个线程完成各自任务:例如应用程序启动类要确保在处理用户 请求前，所有N个外部系统已经启动和运行了。 (3)死锁检测:一个非常方便的使用场景是，你可以使用n个线程访问共享资源，在每次测试阶段的线程数目是不同的，并尝试产生死锁。

相关文章