并发编程—CountDownLatch

发表于 | 分类于 ， |
| 字数统计: 1,395

介绍

同步辅助类
CountDownLatch类位于java.util.concurrent包下，是在jdk5中引入的一个同步辅助类，它允许一个或多个线程去等待其他线程完成一系列操作。

实现原理
CountDownLatch是通过计数器的方式来实现。首先通过CountDownLatch的构造方法给定计数器的初始值，每当在一个线程中调用一次countDown方法，计数器将减去1；CountDownLatch的await方法是会阻塞，直到计数器减至0时await方法则立刻返回。
注意：这个计数器是不能重置的，如果要重置，则需要使用CyclicBarrier类。

CountDownLatch主要方法

1.构造函数

通过传入一个数值(count)来创建一个CountDownLatch，数值表示线程可以从等待状态恢复，countDown方法必须被调用的次数。

1
2
3
4
5
6
7
8
9
10
11
/**
 * Constructs a {@code CountDownLatch} initialized with the given count.
 *
 * @param count the number of times {@link #countDown} must be invoked
 *        before threads can pass through {@link #await}
 * @throws IllegalArgumentException if {@code count} is negative
 */
public CountDownLatch(int count) {
    if (count < 0) throw new IllegalArgumentException("count < 0");
    this.sync = new Sync(count);
}

2.countDown方法

当count本来就为0，此方法不做任何操作，当count比0大，调用此方法进行减1。

1
2
3
4
5
6
7
8
9
10
11
12
13
/**
 * Decrements the count of the latch, releasing all waiting threads if
 * the count reaches zero.
 *
 * <p>If the current count is greater than zero then it is decremented.
 * If the new count is zero then all waiting threads are re-enabled for
 * thread scheduling purposes.
 *
 * <p>If the current count equals zero then nothing happens.
 */
public void countDown() {
    sync.releaseShared(1);
}

3.await方法

1.不带参数
当count比0大，线程会一直等待，直到count的值变为0，或者线程被中断（interepted，此时会抛出中断异常）。

1
2
3
public void await() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
}

2.带参数
当count比0大，线程会等待一段时间，等待时间内如果count的值变为0，返回true；当超出等待时间，返回false；或者等待时间内线程被中断，此时会抛出中断异常。 

1
2
3
4
public boolean await(long timeout, TimeUnit unit)
    throws InterruptedException {
    return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}

应用举例

典型用法1

一个典型应用场景就是启动一个任务时，主线程需要等待多个任务都执行完毕，之后再继续执行后续业务。比如现在需要给1000个手机号发送短信，我把发送短信的任务提交到一个线程池中，等线程池中的发送任务都执行完后，最后再关闭线程池。再比如批量取消订单、批量上传文件等等都是类似的场景。

具体做法是，将CountDownLatch的计数器初始化为new CountDownLatch(n)，每当一个任务线程执行完毕，就将计数器减1 countdownlatch.countDown()，当计数器的值变为0时，在CountDownLatch上 await() 的线程就会被唤醒。

例如需要给1000个手机号发送短信消息，通常可以使用多线程的方式，使用CountDownLatch可以确保给所有手机号都发送了短信，具体做法示例：

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
package com.lzumetal.multithread.counter;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;

/**
 * @author liaosi
 * @date 2018-09-27
 */
public class CountDownTest1 {


    public static void main(String[] args) {
        List<String> mobiles = new ArrayList<>();
        for (int i = 0; i < 1000; i++) {
            mobiles.add(StringUtil.getMobile());
        }

        ExecutorService threadPool = new ThreadPoolExecutor(2, 10,
                3000, TimeUnit.SECONDS, new LinkedBlockingDeque<>());
        CountDownLatch countDownLatch = new CountDownLatch(mobiles.size());
        try {
            for (String mobile : mobiles) {
                threadPool.submit(() -> {
                    try {

                    } catch (Exception e) {
                        sleep();
                        System.out.println(Thread.currentThread().getName() + "发送短信。。。" + mobile);
                    } finally {
                        countDownLatch.countDown();
                    }
                });
            }
            countDownLatch.await();
            System.out.println("===========已全部添加到线程池===========");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            threadPool.shutdown();
        }
    }

    public static void sleep() {
        try {
            TimeUnit.MILLISECONDS.sleep(new Random().nextInt(500));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

典型用法2

另一个个典型用法是：实现多个线程开始执行任务的最大并行性。注意是并行性，不是并发，强调的是多个线程在某一时刻同时开始执行。类似于赛跑，将多个线程放到起点，等待发令枪响，然后同时开跑。

做法是初始化一个共享的CountDownLatch(1)，将其计数器初始化为1，多个线程在开始执行任务前首先 coundownlatch.await()，当主线程调用 countDown() 时，计数器变为0，多个线程同时被唤醒。

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
package com.lzumetal.multithread.counter;

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

/**
 * @author liaosi
 * @date 2018-10-03
 */
public class CountDownTest2 {

    public static void main(String[] args) throws InterruptedException {
        int COUNT = 5;
        CountDownLatch begin = new CountDownLatch(1);
        CountDownLatch end = new CountDownLatch(COUNT);
        ExecutorService executorService = Executors.newFixedThreadPool(COUNT);
        for (int i = 0; i < COUNT; i++) {
            int finalI = i;
            executorService.submit(() -> {
                try {
                    begin.await();
                    System.out.println("运动员" + finalI + "开跑");
                    TimeUnit.SECONDS.sleep(new Random().nextInt(10));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    System.out.println("运动员" + finalI + "已到达终点！");
                    end.countDown();
                }
            });
        }
        System.out.println("倒计时3秒.....");
        TimeUnit.SECONDS.sleep(3);
        begin.countDown();
        System.out.println("比赛开始......");
        end.await();
        System.out.println("比赛结束......");
    }

}

注：仔细分析，上面的起跑应该并不是严格意义上的同时刻并行的。因为虽然线程池的线程数量和任务数量是一样的，这样每一个线程会去执行一个任务，但考虑到cpu的核心数是未知的，并且结合cpu的分片原理，并不能确保5个线程在同一时刻开始执行任务。

------ 本文完 ------