浅析JUC同步工具类之CountDownLatch

前言

谈起 juc(java.util.concurrent)，那就不得不提起一位大神，他就是 Doug Lea，他是 java 并发包的作者，显而易见，他对 java 的影响力也是举足轻重的，在 java 历史上的迭代的过程中，你都可以找到他的身影，其中最重要的便是 JSR-166(Java Specification Requests)的语法标准，而这项标准就来自于 Doug Lea 老爷子所编写的 java.util.concurrent 包。大家可以去http://ifeve.com/doug-lea/ 查看相关译文。

CountDownLatch 概述

image.png
从官方 api doc(译文翻译的不好，谨慎阅读)，可以大致概括一下，CountDownLatch 就是使一个线程在等待(调用 await)另外一个/多个线程完成各自工作之后(调用 countDown)，再继续执行。为了方便理解，我们可以把这个流程理解为”百米赛跑”，我们可以将 CountDownLatch 理解为裁判，裁判在发号施令以后进行等待，而每个运动员在跑到终点以后代表完成任务，当所有的运动员完成以后，裁判统计结果通知所有的运动员成绩。

CountDownLatch 的用法

Methods

image.png
  • await()/await(long timeout,TimeUnit unit)：使当前线程等待锁计数在为 0 之前等待，除非线程被中断或超出了指定的等待时间，如果锁计数为 0，则立即返回。如果锁计数大于 0，则当前线程还是处于等待。
  • countDown()：递减锁计数，如果锁计数到达零，则唤醒等待的线程。如果当前的锁计数大于零，则将计数递减。

总结

  1. 调用 countDown 递减锁计数，如果计数递减为 0，则唤醒调用 await 的等待线程。
  2. 如果调用 await 的当前线程，在调用时已经设置了该线程的中断状态，或者在等待时被中断，则抛出 InterruptedException，并且清除当前线程的中断状态。
  3. 如果超出了指定的 await 等待时间，则立即返回。如果等待时间小于等于 0，则调用 await 的当前线程不会休眠。
  4. 重点：从文档可以看出 countDown 方法并没有限制一个线程只能调用一次，await 方法同样也没有限制。它们会有如下两种场景。
    1. 当同一个线程调用多次 countDown()方法时，每次都会使计数器递减。
    2. 如果多个线程同时执行 await()方法，那么这几个线程都将处于等待状态，并且以共享模式享有同一个锁。

CountDownLatch 的使用场景

image.png
它的使用场景有很多，这里举一个图片识别的例子，我们知道在常见的电商系统中很多商品都有上传图片的功能，通常都是主图跟一些副图批量上传，而当有些监管比较严格的平台，比如政府采购电商平台会将用户上传的所有图片进行一些 OCR 识别，对图片中涉恐涉黄等敏感信息进行监管，但是识别时间会受图片大小、字数、网络环境等等影响浮动，所以为了解决串行识别比较慢的情况下，采用并行识别，然后在通知用户是否图片不合格以提升用户体验。

CountDownLatch 的原理

CountDownLatch 是基于(AQS)AbstractQueuedSynchronizer 实现的，在 AQS 中维护了一个 volatile 类型的整数 state，volatile 可以保证多线程环境下该变量的修改对每个线程都可见，并且由于该属性为整型，因而对该变量的修改也是原子的。创建一个 CountDownLatch 对象时，所传入的整数 n 就会赋值给 state 属性，当 countDown()方法调用时，该线程就会尝试对 state 减一，而调用 await()方法时，当前线程就会判断 state 属性是否为 0，如果为 0，则继续往下执行，如果不为 0，则使当前线程进入等待状态，直到某个线程将 state 属性置为 0，其就会唤醒在 await()方法中等待的线程。

CountDownLatch 使用示例

1、主线程等待子线程执行完成在执行

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

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

/**
 * 主线程等待子线程执行完成再执行
 */
public class CountdownLatchTest1 {
    public static void main(String[] args) {
        ExecutorService service = Executors.newFixedThreadPool(3);
        final CountDownLatch latch = new CountDownLatch(3);
        for (int i = 0; i < 3; i++) {
            Runnable runnable = new Runnable() {
                @Override
                public void run() {
                    try {
                        System.out.println("子线程" + Thread.currentThread().getName() + "开始执行");
                        Thread.sleep((long) (Math.random() * 10000));
                        System.out.println("子线程"+Thread.currentThread().getName()+"执行完成");
                        latch.countDown();//当前线程调用此方法，则计数减一
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
            service.execute(runnable);
        }

        try {
            System.out.println("主线程"+Thread.currentThread().getName()+"等待子线程执行完成...");
            latch.await();//阻塞当前线程，直到计数器的值为0
            System.out.println("主线程"+Thread.currentThread().getName()+"开始执行...");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

2、百米赛跑，4 名运动员选手到达场地等待裁判口令，裁判一声口令，选手听到后同时起跑，当所有选手到达终点，裁判进行汇总排名

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

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

public class CountdownLatchTest2 {
    public static void main(String[] args) {
        ExecutorService service = Executors.newCachedThreadPool();
        final CountDownLatch cdOrder = new CountDownLatch(1);
        final CountDownLatch cdAnswer = new CountDownLatch(4);
        for (int i = 0; i < 4; i++) {
            Runnable runnable = new Runnable() {
                @Override
                public void run() {
                    try {
                        System.out.println("选手" + Thread.currentThread().getName() + "正在等待裁判发布口令");
                        cdOrder.await();
                        System.out.println("选手" + Thread.currentThread().getName() + "已接受裁判口令");
                        Thread.sleep((long) (Math.random() * 10000));
                        System.out.println("选手" + Thread.currentThread().getName() + "到达终点");
                        cdAnswer.countDown();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
            service.execute(runnable);
        }
        try {
            Thread.sleep((long) (Math.random() * 10000));
            System.out.println("裁判"+Thread.currentThread().getName()+"即将发布口令");
            cdOrder.countDown();
            System.out.println("裁判"+Thread.currentThread().getName()+"已发送口令，正在等待所有选手到达终点");
            cdAnswer.await();
            System.out.println("所有选手都到达终点");
            System.out.println("裁判"+Thread.currentThread().getName()+"汇总成绩排名");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        service.shutdown();
    }
}

演示 2 个等待线程通过 CountDownLatch 去等待 3 个工作线程完成操作：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
public class CountDownLatchTest {

    public static void main(String[] args) throws InterruptedException {
        // 让2个线程去等待3个三个工作线程执行完成
        CountDownLatch c = new CountDownLatch(3);

        // 2 个等待线程
        WaitThread waitThread1 = new WaitThread("wait-thread-1", c);
        WaitThread waitThread2 = new WaitThread("wait-thread-2", c);

        // 3个工作线程
        Worker worker1 = new Worker("worker-thread-1", c);
        Worker worker2 = new Worker("worker-thread-2", c);
        Worker worker3 = new Worker("worker-thread-3", c);

        // 启动所有线程
        waitThread1.start();
        waitThread2.start();
        Thread.sleep(1000);
        worker1.start();
        worker2.start();
        worker3.start();
    }
}

/**
 * 等待线程
 */
class WaitThread extends Thread {

    private String name;
    private CountDownLatch c;

    public WaitThread(String name, CountDownLatch c) {
        this.name = name;
        this.c = c;
    }

    @Override
    public void run() {
        try {
            // 等待
            System.out.println(this.name + " wait...");
            c.await();
            System.out.println(this.name + " continue running...");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

/**
 * 工作线程
 */
class Worker extends Thread {

    private String name;
    private CountDownLatch c;

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

    @Override
    public void run() {
        System.out.println(this.name + " is running...");
        try {
            Thread.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(this.name + " is end.");
        c.countDown();
    }
}

CountDownLatch 的不足

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

赏

谢谢你请我吃糖果

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

很惭愧<br><br>只做了一点微小的工作<br>谢谢大家