同步辅助类 CountDownLatch、CyclicBarrier作用

并发编程栏目代码 GitHub package 地址： 点击打开链接

博客并发编程栏目 ： 点击打开链接

对于某种需求，比如 A,B,C 三个并发的线程全部处理完后才执行 D 线程。

可以使用 join 控制等待其他线程完成。

比如多名参赛选手全部准备好后才起跑，即全部到达一个点后执行后面的方法。

Doug Lea 在并发包里封装了更好的处理类

 下面演示CountDownLatch、CyclicBarrier用法

CountDownLatch演示3个线程全部完成后执行后面的方法

package com.thread.concurrent_.basis;

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

/**
* CountDownLatch ：
* 一个同步辅助类，在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待。
* 用给定的计数 初始化 CountDownLatch。
* 由于调用了 countDown()方法，所以在当前计数到达零之前，await方法会一直受阻塞。
* 之后，会释放所有等待的线程，await的所有后续调用都将立即返回。
* 这种现象只出现一次——计数无法被重置。
* <p>
* CountDownLatch 很适合用来将一个任务分为n个独立的部分，等这些部分都完成后继续接下来的任务，
* CountDownLatch 只能出发一次，计数值不能被重置。
*
* @author wei.Li by 14-8-22.
*/
public class CountDownLatchTest {

//同步完成一个事件的执行的线程数量
public static final int SYNCHRONIZED_DONE_THREAD_NUM = 3;

//声明同步辅助类
private static CountDownLatch countDownLatch
= new CountDownLatch(SYNCHRONIZED_DONE_THREAD_NUM);

//线程池
public static final ExecutorService EXECUTOR_SERVICE
= Executors.newFixedThreadPool(SYNCHRONIZED_DONE_THREAD_NUM + 1);

/**
* 模拟多个线程执行任务
*/
public static void analogThreads() {
for (int i = 0; i < SYNCHRONIZED_DONE_THREAD_NUM; i++) {

EXECUTOR_SERVICE.submit(new Callable<Object>() {
@Override
public Object call() throws Exception {

Thread.sleep(new Random().nextInt(20000));
//执行结束后计数器-1
countDownLatch.countDown();

System.out.println(this + " -> done ! " +
".the current countDownLatch is " + countDownLatch.getCount());

return true;
}
});
}
EXECUTOR_SERVICE.shutdown();
}

public static void main(String[] args) {

System.out.println("~~~~~~~~ start ~~~~~~~~");

//模拟多线程协作执行
analogThreads();

try {
//等等计数器归 0 ，即线程全部完成
countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}

System.out.println("~~~~~~~~ done ~~~~~~~~");

}

}

执行结果

~~~~~~~~ start ~~~~~~~~

com.thread.concurrent_.basis.CountDownLatchTest$1@492cf3d0 -> done ! .the current countDownLatch is 2

com.thread.concurrent_.basis.CountDownLatchTest$1@2586b45a -> done ! .the current countDownLatch is 1

com.thread.concurrent_.basis.CountDownLatchTest$1@29911d90 -> done ! .the current countDownLatch is 0

~~~~~~~~ done ~~~~~~~~

CyclicBarrier演示所有选手准备好后才开始起跑

package com.thread.concurrent_.basis;

import java.io.IOException;
import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
* 一个同步辅助类，它允许一组线程互相等待，直到到达某个公共屏障点 (common barrier point)。
* 在涉及一组固定大小的线程的程序中，这些线程必须不时地互相等待，此时 CyclicBarrier 很有用。
* 因为该 barrier 在释放等待线程后可以重用，所以称它为循环 的 barrier。
* <p>
* 需要所有的子任务都完成时，才执行主任务，这个时候就可以选择使用CyclicBarrier。
* <p>
* 演示所有选手进入场地后，全部准备好后开始起跑！
*
* @author wei.Li by 14-8-23.
*/
public class CyclicBarrierTest {

//参加跑步比赛的人数
public static final int FOOTRACE_NUM = 5;

//线程池
private static ExecutorService executor
= Executors.newFixedThreadPool(FOOTRACE_NUM);

private static CyclicBarrier barrier
= new CyclicBarrier(FOOTRACE_NUM);

// 一个同步辅助类，它允许一组线程互相等待，直到到达某个公共屏障点 (common barrier point)
static class Footrace implements Runnable {

private CyclicBarrier barrier;

private String name;

public Footrace(CyclicBarrier barrier, String name) {
super();
this.barrier = barrier;
this.name = name;
}

@Override
public void run() {
try {

System.out.println(name + "  in the location...");
Thread.sleep(1000 * (new Random()).nextInt(8));//准备中
System.out.println(name + "  say : I am ready ...");

// barrier的await方法，在所有参与者都已经在此 barrier 上调用 await 方法之前，将一直等待。
barrier.await();
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
System.out.println(name + " go ！");
}
}

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

for (int i = 0; i < FOOTRACE_NUM; i++) {
executor.submit(new Thread(
new Footrace(barrier, i + "号选手")
));
}
executor.shutdown();
}
}

执行结果

0号选手  in the location...

2号选手  in the location...

3号选手  in the location...

1号选手  in the location...

4号选手  in the location...

2号选手  say : I am ready ...

3号选手  say : I am ready ...

4号选手  say : I am ready ...

1号选手  say : I am ready ...

0号选手  say : I am ready ...

0号选手 go ！

2号选手 go ！

3号选手 go ！

1号选手 go ！

4号选手 go ！