并发编程常用工具类之countDownLatch和cyclicBarrier的使用对比

Posted 2020-12-30 gmt-hao

1.CountDownLatch

          countDownLatch的作用是让一组线程等待其他线程完成工作以后在执行，相当于加强版的join(不懂可以百度一下join的用法)，一般在初始化的时候会在构造方法传入计数器，

后续，在其他线程中每次调用countDown方法计数器减一，一般在需要等待的线程中调用countDownLatch的await方法阻塞线程，在当计数器为0时，等待线程继续运行。

          光看上面的定义描述不是很直观，我们再来结合代码看一下实际运用：

 1 public class UseCountDownLatch {
 2     
 3     static CountDownLatch latch = new CountDownLatch(6);
 4     //初始化线程(只有一步，有4个)
 5     private static class InitThread implements Runnable{
 6 
 7         @Override
 8         public void run() {
 9             System.out.println("Thread_"+Thread.currentThread().getId()
10                     +" ready init work......");
11             latch.countDown();//初始化线程完成工作了，countDown方法只扣减一次；
12             for(int i =0;i<2;i++) {
13                 System.out.println("Thread_"+Thread.currentThread().getId()
14                         +" ........continue do its work");
15             }
16         }
17     }
18     //业务线程
19     private static class BusiThread implements Runnable{
20 
21         @Override
22         public void run() {
23             try {
24                 latch.await();
25             } catch (InterruptedException e) {
26                 e.printStackTrace();
27             }
28             for(int i =0;i<3;i++) {
29                 System.out.println("BusiThread_"+Thread.currentThread().getId()
30                         +" do business-----");
31             }
32         }
33     }
34 
35     public static void main(String[] args) throws InterruptedException {
36         //单独的初始化线程,初始化分为2步，需要扣减两次
37         new Thread(new Runnable() {
38             @Override
39             public void run() {
40                 SleepTools.ms(1);
41                 System.out.println("Thread_"+Thread.currentThread().getId()
42                         +" ready init work step 1st......");
43                 latch.countDown();//每完成一步初始化工作，扣减一次
44                 System.out.println("begin step 2nd.......");
45                 SleepTools.ms(1);
46                 System.out.println("Thread_"+Thread.currentThread().getId()
47                         +" ready init work step 2nd......");
48                 latch.countDown();//每完成一步初始化工作，扣减一次
49             } 
50         }).start();
51         new Thread(new BusiThread()).start();
52         for(int i=0;i<=3;i++){
53             Thread thread = new Thread(new InitThread());
54             thread.start();
55         }
56 
57         latch.await();
58         System.out.println("Main do ites work........");
59     }
60 }

 

运行结果可以看到，两个初始化线程先跑，当两个初始化线程跑完了，latch的计数器减为0，阻塞放开，主线程和业务线程继续往下运行。但是，在设计这部分算法的时候需要注意，有可能会出现计数器没有减为0，则线程一直阻塞，导致程序卡死。

          countDownLatch一般使用在多线程并发之后需要对结果进行处理，而我们无法控制所有线程的执行时间，所以在这里加上阻塞，等到只有线程全部执行完。

 

2.CyclicBarrier

          cyclicBarrier从业务角度来说，和countDownLatch比较类似(具体对比后面会专门介绍)，其作用类似一个屏障(英文中也是屏障的意思)，阻隔线程直到全部到达屏障点，放开屏障。一般可以用在多线程统计的时候，

          从代码角度来看，cyclicBarrier有两个构造方法，CyclicBarrier(int parties)和CyclicBarrier(int parties, Runnable barrierAction)；

          两个方法的区别是第一个只是记录了线程计数器的个数，而第二个不仅记录了计数器，当屏障放开时，会执行第二个参数线程的方法(第二个参数一般传入的是一个实现了Runnable的线程方法)

再来结合代码深入了解一下：

          

 1 public class UseCyclicBarrier {
 2     
 3     private static CyclicBarrier barrier 
 4         = new CyclicBarrier(5,new CollectThread());
 5     
 6     private static ConcurrentHashMap<String,Long> resultMap
 7             = new ConcurrentHashMap<>();//存放子线程工作结果的容器
 8 
 9     public static void main(String[] args) {
10         new Thread(){
11             @Override
12             public void run() {
13                 long id = Thread.currentThread().getId();//线程本身的处理结果
14                 resultMap.put(Thread.currentThread().getId()+"",id);
15                 Random r = new Random();//随机决定工作线程的是否睡眠
16                 try {
17                     if(r.nextBoolean()) {
18                         Thread.sleep(2000+id);
19                         System.out.println("Thread_"+id+" ....do something ");
20                     }
21                     System.out.println(id+"....is await");
22                     barrier.await();
23                     Thread.sleep(1000+id);
24                     System.out.println("Thread_"+id+" ....do its business ");
25                 } catch (Exception e) {
26                     e.printStackTrace();
27                 }
28             }
29         }.start();
30         for(int i=0;i<=3;i++){
31             Thread thread = new Thread(new SubThread());
32             thread.start();
33         }
34     }
35 
36     //负责屏障开放以后的工作
37     private static class CollectThread implements Runnable{
38 
39         @Override
40         public void run() {
41             StringBuilder result = new StringBuilder();
42             for(Map.Entry<String,Long> workResult:resultMap.entrySet()){
43                 result.append("["+workResult.getValue()+"]");
44             }
45             System.out.println(" the result = "+ result);
46             System.out.println("do other business........");
47         }
48     }
49 
50     //工作线程
51     private static class SubThread implements Runnable{
52 
53         @Override
54         public void run() {
55             long id = Thread.currentThread().getId();//线程本身的处理结果
56             resultMap.put(Thread.currentThread().getId()+"",id);
57             Random r = new Random();//随机决定工作线程的是否睡眠
58             try {
59                 if(r.nextBoolean()) {
60                     Thread.sleep(2000+id);
61                     System.out.println("Thread_"+id+" ....do something ");
62                 }
63                 System.out.println(id+"....is await");
64                 barrier.await();
65                 Thread.sleep(1000+id);
66                 System.out.println("Thread_"+id+" ....do its business ");
67             } catch (Exception e) {
68                 e.printStackTrace();
69             }
70         }
71     }
72 }

   输出结果：       11....is await
                            12....is await
                             Thread_10 ....do something
                     10....is await
Thread_13 ....do something
13....is await
Thread_14 ....do something
14....is await
 the result = [11][12][13][14][10]
do other business........
Thread_10 ....do its business
Thread_11 ....do its business
Thread_12 ....do its business
Thread_13 ....do its business
Thread_14 ....do its business

 

3.countDownLatch和cyclicBarrier区别

        （1）countDownlatch的计数器由调用countDown方法次数决定，每次调用计数器减一，可以在一个线程中调用多次，而cyclicBarrier计数器取决调用await方法的线程个数。

          (2)  countDownLatch只能用一次，而cyclicBarrier可以循环使用。并且cyclicBarrier可以调用reset方法重置计数器，可以在线程故障重新启用线程调用。

          (3)  二者在内部方法也有很多区别，具体有兴趣的可以去看看源码。