CountDownLatch和CyclicBarrier
使用场景
场景一
工作中需要在web服务对外之前，spring的bean初始化的时候加载数据到缓存中。
但是由于数据量过大，需要多线程加载。要求所有的缓存加载成功之后，这个bean才初始化成功，程序继续往下走。
场景二
商务合作的需求，需要显示的数据来自合作方和我们自己的数据库中。
之前是只显示我们数据库点数据，现在合作方提供了一个接口让我实时调用对方的接口，并把两部分的数据合并后返回给前端。
但是由于合作方的接口不是特别稳定，而且也不能保证高可用，所以可以考虑同时从我们和合作方取数据，设置超时时间，如果都返回了数据就合并给前端，如果对方未能返回数据，还是有我们自己的数据能显示给用户的。
CountDownLatch和CyclicBarrier
综合上述两个场景，我看了CountDownLatch和CyclicBarrier，看说明觉得比较适合我的使用。
CountDownLatch
package countdownlatchtest;
import com.google.common.collect.Maps;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
public class CountDownLatchService {
private CountDownLatch countDownLatch = new CountDownLatch(4);
/**
* 用来存储所有线程的运行结果
*/
private ConcurrentMap<String, String> resultMap = Maps.newConcurrentMap();
public CountDownLatch getCountDownLatch() {
return countDownLatch;
}
public ConcurrentMap<String, String> getResultMap() {
return resultMap;
}
}
package countdownlatchtest;
import org.apache.commons.lang3.RandomUtils;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
public class Worker implements Runnable {
private ConcurrentMap<String, String> map;
private CountDownLatch countDownLatch;
public Worker(ConcurrentMap<String, String> map, CountDownLatch countDownLatch) {
this.map = map;
this.countDownLatch = countDownLatch;
}
@Override
public void run() {
// 这里写代码做某些事
System.out.println(Thread.currentThread().getName() + "\t开始了...");
final int sleep = RandomUtils.nextInt(5, 20);
try {
Thread.sleep(sleep * 1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 添加结果到map中
map.putIfAbsent(Thread.currentThread().getName(), String.valueOf(sleep));
// 告诉 countDownLatch ，当前线程完成了
System.out.println(Thread.currentThread().getName() + "\t结束了...");
countDownLatch.countDown();
}
}
package countdownlatchtest;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class Main {
public static void main(String[] args) {
final ExecutorService executorService = Executors.newFixedThreadPool(6);
for (int i = 0; i < 3; i++) {
// 模拟多次执行任务
doTask(executorService);
}
}
private static void doTask(ExecutorService executorService) {
CountDownLatchService countDownLatchService = new CountDownLatchService();
for (int i = 0; i < 4; i++) {
Worker worker = new Worker(countDownLatchService.getResultMap(), countDownLatchService.getCountDownLatch());
executorService.submit(worker);
}
try {
// 阻塞在这里，等待其他线程执行完成
countDownLatchService.getCountDownLatch().await();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("所有线程执行成功");
System.out.println("打印结果如下：\n" + countDownLatchService.getResultMap());
}
}
CyclicBarrier
package cyclicbarriertest;
import com.google.common.collect.Maps;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CyclicBarrier;
public class CyclicBarrierService implements Runnable {
/**
* 表示当有4个线程执行完的时候，会调用第二个参数 barrierAction 的start()方法
*
* 所以本类实现 Runnable 接口，传入this
*/
private CyclicBarrier cyclicBarrier = new CyclicBarrier(4, this);
/**
* 用来存储所有线程的运行结果
*/
private ConcurrentMap<String, String> resultMap = Maps.newConcurrentMap();
@Override
public void run() {
System.out.println("所有线程执行成功");
System.out.println("打印结果如下：\n" + resultMap);
}
public ConcurrentMap<String, String> getResultMap() {
return resultMap;
}
public CyclicBarrier getCyclicBarrier() {
return cyclicBarrier;
}
}
package cyclicbarriertest;
import org.apache.commons.lang3.RandomUtils;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CyclicBarrier;
public class Worker implements Runnable {
private ConcurrentMap<String, String> map;
private CyclicBarrier cyclicBarrier;
public Worker(ConcurrentMap<String, String> map, CyclicBarrier cyclicBarrier) {
this.map = map;
this.cyclicBarrier = cyclicBarrier;
}
@Override
public void run() {
// 这里写代码做某些事
System.out.println(Thread.currentThread().getName() + "\t开始了...");
final int sleep = RandomUtils.nextInt(5, 20);
try {
Thread.sleep(sleep * 1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 添加结果到map中
map.putIfAbsent(Thread.currentThread().getName(), String.valueOf(sleep));
try {
// 告诉 cyclicBarrier ，当前线程完成了
System.out.println(Thread.currentThread().getName() + "\t结束了...");
cyclicBarrier.await();
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
}
}
package cyclicbarriertest;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class Main {
public static void main(String[] args) {
final ExecutorService executorService = Executors.newFixedThreadPool(6);
for (int j = 0; j < 3; j++) {
// 模拟多次任务执行
doTask(executorService);
}
}
private static void doTask(ExecutorService executorService) {
CyclicBarrierService cyclicBarrierService = new CyclicBarrierService();
for (int i = 0; i < 4; i++) {
Worker worker = new Worker(cyclicBarrierService.getResultMap(), cyclicBarrierService.getCyclicBarrier());
executorService.submit(worker);
}
}
}
总结
写了上面两个例子，后面发现场景一适合用CountDownLatch，场景二适合用CyclicBarrier。场景一我们基本上不存在有任务不能执行完的情况，基本上做到计数器不归0，即使服务启动了也没办法正常使用，场景二很多情况都是任务不能正常的执行完成。