并发编程——闭锁CountDownLatch 与栅栏CyclicBarrier

时间：2014-09-09 15:31:18 阅读：346 评论：0 收藏：0 [点我收藏+]

标签：style blog http color os io 使用 java ar

Java并发编程实践目录

并发编程—— ConcurrentHashMap

并发编程—— 阻塞队列和生产者-消费者模式

并发编程——闭锁CountDownLatch 与栅栏CyclicBarrier

概述

第1部分闭锁

第2部分栅栏

参考

第1部分闭锁

　　闭锁是一种同步工具类，可以延迟线程的进度直到其到达终止状态。闭锁的作用相当于一扇门：在闭锁到达结束状态之前，这扇门一直是关闭的，并且没有任何线程能通过，当到达结束状态时，这扇门会打开并允许所有的线程通过。当闭锁到达结束状态后，将不会再改变状态，因此这扇门将永远保持打开状态。闭锁可以用来确保某些活动直到其他活动都完成后才继续执行。例如：

确保某个计算在其需要的所有资源都被初始化之后才继续执行。
确保某个服务在其依赖的所有其他服务都已经启动之后才启动。
等待直到某个操作的所有参与者都就绪再继续执行。

　　CountDownLatch是一种灵活的闭锁实现，可以再上述各种情况中使用，它可以使一个或多个线程等待一组事件发生。闭锁状态包括一个计数器，该计数器被初始化为一个正数，表示需要等待的事件数量。countDown方法递减计数器，表示有一个事件已经发生了，而await方法等待计数器达到0，这表示所有需要等待的事件都已经发生。如果计数器的值非零，那么await会一直阻塞直到计数器为零，或者等待中的线程中断，或者等待超时。

　　下面程序TestHarness中给出了闭锁的两种常见用法。TestHarness创建一定数量的线程，利用它们并发地执行指定的任务。它使用两个闭锁，分别表示“起始门”和“结束门”。起始门计数器的初始值为1，而结束门计数器的初始值为工作线程的数量。每个工作线程首先要做到就是在启动门上等待，从而确保所有线程都就绪后才开始执行。而每个线程要做的最后一个事情是将调用结束门的countDown方法减1 ，这能使主线程高效低等待直到所有工作线程都执行完毕，因此可以统计所消耗的时间。

 1 package com.concurrency.BasicBuildingBlocks_5;
 2 
 3 import java.util.concurrent.CountDownLatch;
 4 
 5 /**
 6  * 5.11 在计时测试中使用CountDownLatch来启动和停止线程（闭锁）
 7  * 
 8  * @ClassName: TestHarness
 9  * TODO
10  * @author Xingle
11  * @date 2014-9-4 下午2:56:29
12  */
13 public class TestHarness {
14     
15     int nThreads ;
16     Runnable task;
17     public TestHarness(int nThreads,Runnable task){
18         this.nThreads = nThreads;
19         this.task = task;
20     } 
21      public long timeTask(){
22          
23         //起始门
24         final CountDownLatch startGate = new CountDownLatch(1);
25         //结束门
26         final CountDownLatch endGate = new CountDownLatch(nThreads);
27          for(int i = 0;i<nThreads;i++){
28              Thread thread = new Thread(){
29                  public void run(){
30                      //每个线程在启动门上等待，确保所有线程都就绪后才开始
31                      try {
32                         startGate.await();//等待计数器达到0 
33                         try{
34                             task.run();
35                         }finally{
36                             //每个线程结束后，调用countDown递减计数器，表示一个事件发生
37                             endGate.countDown();
38                         }
39                     } catch (InterruptedException e) {
40                         e.printStackTrace();
41                     }
42                      
43                  }
44              };
45              thread.start();
46          }
47          long start = System.nanoTime();
48          //启动门发生
49          startGate.countDown();
50          try {
51              //等待结束门的线程都结束
52             endGate.await();
53         } catch (InterruptedException e) {
54             e.printStackTrace();
55         }
56         long end = System.nanoTime();
57         return end - start;
58      }
59 }

测试程序：

 1 package com.concurrency.BasicBuildingBlocks_5;
 2 
 3 /**
 4  * 
 5  * @ClassName: TestHarnessMain
 6  * TODO
 7  * @author Xingle
 8  * @date 2014-9-4 下午3:27:18
 9  */
10 public class TestHarnessMain {
11 
12     public static void main(String[] args){
13         Runnable task = new Runnable() {
14             
15             @Override
16             public void run() {
17                 System.out.println("执行任务，我是线程："+Thread.currentThread().getName());                
18             }
19         };
20         int count = 10;
21         TestHarness testHarness = new TestHarness(count, task );
22         long time = testHarness.timeTask();
23         System.out.println("闭锁  测试结果  执行"+count+"个线程"+" 一共用时："+time);
24     }
25 }

执行结果：

bubuko.com,布布扣

第2部分栅栏

上面已经看到通过闭锁来启动一组相关的操作，或者等待一组相关的操作结束。闭锁是一次性对象，一旦进入终止状态，就不能被重置。

　　栅栏（Barrier）类似于闭锁，它能阻塞一组线程直到某个事件发生。栅栏与闭锁的关键区别在于，所有线程必须同时到达栅栏位置，才能继续执行。闭锁用于等待事件，而栅栏用于等待其他线程。（栅栏则是所有线程相互等待，直到所有线程都到达某一点时才打开栅栏，然后线程可以继续执行。）

　　CyclicBarrier 可以使一定数量的参与方反复地在栅栏位置汇集，它在并行迭代算法中非常有用。CyclicBarrier支持一个可选的 Runnable 参数，当线程通过栅栏时，runnable对象将被调用。构造函数CyclicBarrier(int parties, Runnable barrierAction)，当线程在CyclicBarrier对象上调用await()方法时，栅栏的计数器将增加1，当计数器为parties时，栅栏将打开。

 1 package com.concurrency.BasicBuildingBlocks_5;
 2 
 3 import java.util.concurrent.BrokenBarrierException;
 4 import java.util.concurrent.CyclicBarrier;
 5 
 6 /**
 7  * 
 8  * @ClassName: Worker 
 9  * @author Xingle
10  * @date 2014-9-9 上午10:41:17
11  */
12 public class Worker implements Runnable {
13 
14     final int id;
15     final CyclicBarrier barrier;
16 
17     public Worker(int id, CyclicBarrier barrier) {
18         this.id = id;
19         this.barrier = barrier;
20     }
21 
22     @Override
23     public void run() {
24         System.out.println(this.id + " start to run!");
25         try {
26             this.barrier.await();
27         } catch (InterruptedException e) {
28             e.printStackTrace();
29         } catch (BrokenBarrierException e) {
30             e.printStackTrace();
31         }
32 
33     }
34 
35 }

测试程序：

 1 package com.concurrency.BasicBuildingBlocks_5;
 2 
 3 import java.util.concurrent.CyclicBarrier;
 4 
 5 /**
 6  * 
 7  * @ClassName: Beer
 8  * 有五个人参与跑步，规定五个人只要都跑到终点了，大家可以喝啤酒。但是，只要有一个人没到终点，就不能喝。 这里没有要求大家要同时起跑
 9  * @author Xingle
10  * @date 2014-9-9 上午10:40:36
11  */
12 public class Beer {
13     public static void main(String[] args){
14         final int count = 5;
15         final CyclicBarrier barrier = new CyclicBarrier(count, new Runnable() {
16             
17             @Override
18             public void run() {
19                 System.out.println("drink beer!");
20             }
21         });
22         
23         for (int i =0; i<count;i++){
24             new Thread(new Worker(i, barrier)).start();
25         }
26     }
27     
28 }