CountDownLatch 的用法

2023-12-21 11:53:15

CountDownLatch 的用法

CountDownLatch是一个同步工具类，它使用给定的 count初始化， await()方法会一直阻塞，直到计数器的值变为零（由于 countDown()方法被调用导致的），这时会释放所有等待的线程，且之后再调用 await()方法会直接返回，不会阻塞。 CountDownLatch是一个一次性的类，计数器不能被重置，这一点与 CyclicBarrier不同。另一个不同点是： CountDownLatch是让所有线程等待计数器的值变为零再继续执行；而 CyclicBarrier是要等待指定个数的线程到达 Barrier 的位置再一起继续执行。

方法

构造方法 CountDownLatch(int count)

计数器的初始值为count，也就是说countDown()方法至少被调用count次等待的线程才会被唤醒。如果count为负数或抛出异常IllegalArgumentException。

countDown()

如果当前计数器的值大于零，则将其减一，如果新的计数器值等于零，则释放所有等待的线程。
如果当前计数器为零，则什么都不做。
此方法不会阻塞。

long getCount()

获取当前计数器的值。

public static void test() throws InterruptedException {
	CountDownLatch cdl = new CountDownLatch(3);
	System.out.println(cdl.getCount());
	cdl.countDown();
	System.out.println(cdl.getCount());
	cdl.countDown();
	System.out.println(cdl.getCount());
	cdl.countDown();
	System.out.println(cdl.getCount());
	cdl.countDown();
	System.out.println(cdl.getCount());
	cdl.countDown();
	System.out.println(cdl.getCount());
}

计数器变为零后就不会再减了。

await()

导致当前线程等待，直到计数器的值变为零，除非线程被中断。如果计数器已经为零了，则立即返回。以下两种情况会抛出InterruptedException并清空中断标志：

在调用wait()方法前，当前线程的中断状态已经为 true 了
在等待的过程中被中断了

模拟两种中断情况

// 在调用`wait()`方法前，当前线程的中断状态已经为 true 了
public static void test1() throws InterruptedException {
	CountDownLatch cdl = new CountDownLatch(1);
	Thread.currentThread().interrupt();
	cdl.await();
}

// 在等待的过程中被中断了
public static void test2() throws InterruptedException {
	CountDownLatch cdl = new CountDownLatch(1);
	Thread t1 = new Thread(() -> {
		try {
			cdl.await();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}, "t1");
	t1.start();
	Thread.sleep(500);
	t1.interrupt();
}

boolean await(long timeout, TimeUnit unit)

此方法与await()的不同点：

此方法至多会等待指定的时间，超时后会自动唤醒，若 timeout 小于等于零，则不会等待
次方法有 boolean 类型的返回值：若计数器变为零了，则返回 true；若指定的等待时间过去了，则返回 false

等待指定时间

public static void test3() throws InterruptedException {
	CountDownLatch cdl = new CountDownLatch(1);
	log.info("开始 await");
	boolean b = cdl.await(2, TimeUnit.SECONDS);
	log.info("结束 await, 返回值: {}", b);
}

等待 2 秒后返回了，且返回值为 false

16:33:11.492 [main] INFO com.example.heima.concurrent.CountDownLatchTest - 开始 await
16:33:13.503 [main] INFO com.example.heima.concurrent.CountDownLatchTest - 结束 await, 返回值: false

计数器在等待过程中变为零

public static void test4() throws InterruptedException {
	CountDownLatch cdl = new CountDownLatch(1);
	Thread t1 = new Thread(() -> {
		try {
			log.info("开始 await");
			// 至多等待 2 秒
			boolean b = cdl.await(2, TimeUnit.SECONDS);
			log.info("结束 await, 返回值: {}", b);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}, "t1");
	t1.start();

	Thread.sleep(1000);
	cdl.countDown();
}

等待 1 秒后返回了，且返回值为 true

16:36:20.408 [t1] INFO com.example.heima.concurrent.CountDownLatchTest - 开始 await
16:36:21.421 [t1] INFO com.example.heima.concurrent.CountDownLatchTest - 结束 await, 返回值: true

计数器在调用await()方法前就变为 0 了

public static void test5() throws InterruptedException {
	CountDownLatch cdl = new CountDownLatch(1);
	cdl.countDown();
	log.info("开始 await");
	boolean b = cdl.await(2, TimeUnit.SECONDS);
	log.info("结束 await, 返回值: {}", b);
}

不会等待，且返回值为 true

16:38:33.047 [main] INFO com.example.heima.concurrent.CountDownLatchTest - 开始 await
16:38:33.054 [main] INFO com.example.heima.concurrent.CountDownLatchTest - 结束 await, 返回值: true

两个示例

以下代码均来源于源码的注释
Driver、Worker
下面是两个类，其中一组 Worker 线程使用了两个CountDownLatch：

第一个是启动信号，阻止任何 Worker 继续，直到 Driver 让他们继续
第二个是完成信号，它允许 Driver 等待所有的 Worker 完成

class Driver { // ...
    public static void main(String[] args) throws InterruptedException {
        int N = 5;
        CountDownLatch startSignal = new CountDownLatch(1);
        CountDownLatch doneSignal = new CountDownLatch(N);

        for (int i = 0; i < N; ++i) // create and start threads
            new Thread(new Worker(startSignal, doneSignal)).start();

        System.out.println("doSomethingElse");          // don't let run yet
        startSignal.countDown();                        // let all threads proceed
        System.out.println("doSomethingElse");
        doneSignal.await();                             // wait for all to finish
        System.out.println("all worker completed");
    }
}

class Worker implements Runnable {
    private final CountDownLatch startSignal;
    private final CountDownLatch doneSignal;
    Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {
        this.startSignal = startSignal;
        this.doneSignal = doneSignal;
    }
    public void run() {
        try {
            startSignal.await();
            doWork();
            doneSignal.countDown();
        } catch (InterruptedException ex) {} // return;
    }

    void doWork() {
        System.out.println("doWork");
    }
}

将一个问题分解为多个部分

class Driver2 { // ...
    public static void main(String[] args) throws InterruptedException {
        int N = 3;
        CountDownLatch doneSignal = new CountDownLatch(N);
        Executor e = Executors.newFixedThreadPool(N);
		
		// i 代表是问题的第几部分
        for (int i = 0; i < N; ++i) // create and start threads
            e.execute(new WorkerRunnable(doneSignal, i));

        doneSignal.await();           // wait for all to finish
        System.out.println("all task completed");
    }
}

class WorkerRunnable implements Runnable {
    private final CountDownLatch doneSignal;
    private final int i;
    WorkerRunnable(CountDownLatch doneSignal, int i) {
        this.doneSignal = doneSignal;
        this.i = i;
    }
    public void run() {
        doWork(i);
        doneSignal.countDown();
    }

    void doWork(int i) {
        System.out.println("task " + i);
    }
}

码农公寓