CyclicBarrier 和CountDownLatch使用详解

http://xijunhu.iteye.com/blog/713433

http://www.blogjava.net/jlins-you/archive/2012/04/24/376516.html

徐明明：http://xumingming.sinaapp.com/215/countdownlatch-vs-cyclicbarrier/

CountDownLatch : 一个线程(或者多个)，等待另外N个线程完成某个事情之后才能执行。

CyclicBarrier : N个线程相互等待，任何一个线程完成之前，所有的线程都必须等待。
这样应该就清楚一点了，对于CountDownLatch来说，重点是那个“一个线程”, 是它在等待，而另外那N的线程在把“某个事情”做完之后可以继续等待，可以终止。而对于CyclicBarrier来说，重点是那N个线程，他们之间任何一个没有完成，所有的线程都必须等待。

CountDownLatch 是计数器, 线程完成一个就记一个, 就像报数一样, 只不过是递减的.

CyclicBarrier像是一个（主线程）裁判，其他子线程是绕圈跑的运动员。每一轮都得等到所有运动员都跑到了终点（也是起点），裁判宣读下本轮的结果，然后再发令，运动员开始下一轮的绕圈跑。当然，裁判也可以没有任何动作，等所有运动员都到终点后，就直接发令开跑。

class Worker implements Runnable {CyclicBarrier barrier;int id;int epoch;public Worker(CyclicBarrier barrier, int id, int epoch) {this.barrier = barrier;this.id = id;this.epoch = epoch;}@Overridepublic void run() {for (int i = 0; i < epoch; ++i) {System.out.println("epoch:"+i+ " worker " + id + " start");try {barrier.await();} catch (BrokenBarrierException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}System.out.println("epoch:"+i+ " worker " + id + " end");try {TimeUnit.MICROSECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}}}}public static void main(String[] args) throws Exception{int n = 3;CyclicBarrier barrier = new CyclicBarrier(n, new Runnable() {@Overridepublic void run() {try {TimeUnit.SECONDS.sleep(3);System.out.println("Master: all workers had come, now go!");} catch (InterruptedException e) {e.printStackTrace();}}});for (int i = 0; i < n; ++i)new Thread(new Worker(barrier, i, 3)).start();}

output:

epoch:0 worker 1 start
epoch:0 worker 0 start
epoch:0 worker 2 start
Master: all workers had come, now go!
epoch:0 worker 1 end
epoch:0 worker 2 end
epoch:0 worker 0 end
epoch:1 worker 1 start
epoch:1 worker 0 start
epoch:1 worker 2 start
Master: all workers had come, now go!
epoch:1 worker 1 end
epoch:1 worker 2 end
epoch:1 worker 0 end
epoch:2 worker 0 start
epoch:2 worker 1 start
epoch:2 worker 2 start
Master: all workers had come, now go!
epoch:2 worker 0 end
epoch:2 worker 2 end
epoch:2 worker 1 end

http://blog.csdn.net/yanhandle/article/details/9016329

CountDownLatch举例

CountDownLatch是一个计数器闭锁，主要的功能就是通过await()方法来阻塞住当前线程，然后等待计数器减少到0了，再唤起这些线程继续执行。这个类里主要有两个方法，一个是向下减计数器的方法:countdown(),其实现的核心代码如下:

public boolean tryReleaseShared(int releases) {  // Decrement count; signal when transition to zero  for (;;) {   int c = getState();     if (c == 0)    return false;     int nextc = c-1;    if (compareAndSetState(c, nextc))    return nextc == 0;    }
}

很简单，如果取得当前的状态为0,说明这个锁已经结束，直接返回false;如果没有结束，然后去设置计数器减1,如果compareAndSetState不成功，则继续循环执行。而其中的一直等待计数器归零的方法是await()。
通过CountDownLatch可以做几件事情：

1. 主线程控制同时启动一组线程

final CountDownLatch count = new CountDownLatch(1);
for (int i = 0; i < 3; i++) {new Thread("Thread" + i) {public void run() {System.out.println(Thread.currentThread().getName() + " wait");try {count.await();} catch (InterruptedException e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName() + " start");}}.start();
}
//等等三秒，否则有可能3个线程并没有全部进行await状态
try {Thread.sleep(3000);
} catch (InterruptedException e) {e.printStackTrace();
}
count.countDown();

2. 主线程等待各子线程全部执行完毕后再往下执行:

final CountDownLatch count = new CountDownLatch(3);
for (int i = 0; i < 3; i++) {new Thread("Thread" + i) {public void run() {System.out.println(Thread.currentThread().getName() + " start");count.countDown();}}.start();
}
try {count.await();
} catch (InterruptedException e) {e.printStackTrace();
}
System.out.println("All end!!!");

import java.util.Random;
import java.util.concurrent.CyclicBarrier;/** *//*** CyclicBarrier类似于CountDownLatch也是个计数器，* 不同的是CyclicBarrier数的是调用了CyclicBarrier.await()进入等待的线程数，* 当线程数达到了CyclicBarrier初始时规定的数目时，所有进入等待状态的线程被唤醒并继续。* CyclicBarrier就象它名字的意思一样，可看成是个障碍，* 所有的线程必须到齐后才能一起通过这个障碍。* CyclicBarrier初始时还可带一个Runnable的参数，* 此Runnable任务在CyclicBarrier的数目达到后，所有其它线程被唤醒前被执行。*/
public class CyclicBarrierTest {public static class ComponentThread implements Runnable {CyclicBarrier barrier;// 计数器int ID;    // 组件标识int[] array;    // 数据数组// 构造方法public ComponentThread(CyclicBarrier barrier, int[] array, int ID) {this.barrier = barrier;this.ID = ID;this.array = array;}public void run() {try {array[ID] = new Random().nextInt(100);System.out.println("Component " + ID + " generates: " + array[ID]);// 在这里等待Barrier处System.out.println("Component " + ID + " sleep");barrier.await();System.out.println("Component " + ID + " awaked");// 计算数据数组中的当前值和后续值int result = array[ID] + array[ID + 1];System.out.println("Component " + ID + " result: " + result);} catch (Exception ex) {}}}/** *//*** 测试CyclicBarrier的用法*/public static void testCyclicBarrier() {final int[] array = new int[3];CyclicBarrier barrier = new CyclicBarrier(2, new Runnable() {// 在所有线程都到达Barrier时执行public void run() {System.out.println("testCyclicBarrier run");array[2] = array[0] + array[1];}});// 启动线程new Thread(new ComponentThread(barrier, array, 0)).start();new Thread(new ComponentThread(barrier, array, 1)).start();}public static void main(String[] args) {CyclicBarrierTest.testCyclicBarrier();}
}

说明：在main中执行testCyclicBarrier方法
执行到CyclicBarrier barrier = new CyclicBarrier(2, new Runnable()...)时
Runnable的参数是在CyclicBarrier的数目达到2时并且所有被CyclicBarrier.await()进入等待的线程被唤醒前执行。
所以继续执行下面的两个线程
new Thread(new ComponentThread(barrier, array, 0)).start();
new Thread(new ComponentThread(barrier, array, 1)).start();
执行public void run()方法，分别执行，互不影响
执行到barrier.await();时该线程进入等待状态，当两个线程都执行完barrier.await();时，进入到new CyclicBarrier(2, new Runnable()...)里面的方法，执行完里面的方法后，等待的两个线程再次被唤醒，继续各自执行线程后面的语句。

首先在cyclicBarrier中设置线程数目n,

各个线程执行到barrier.await进入等待状态，

当各个线程调用cyclicBarrier中await的方法的次数达到n次时，

开始执行cyclicBarrier中定义的线程方法，待执行结束之后，等待的n个线程再次被唤醒，继续执行await之后的语句。

也就是说对于cyclicbarrier而言，不知道执行各个线程是啥，只知道一旦自己的await被调用了n次，自己内部的线程就开始执行，执行完之后，被堵塞的n个线程才继续执行

CyclicBarrier 支持一个可选的 Runnable 命令，在一组线程中的最后一个线程到达之后（但在释放所有线程之前），该命令只在每个屏障点运行一次。
若在继续所有参与线程之前更新共享状态，此屏障操作很有用。

示例用法：下面是一个在并行分解设计中使用 barrier 的例子：

class Solver {
   final int N;
   final float[][] data;
   final CyclicBarrier barrier;

   class Worker implements Runnable {
     int myRow;
     Worker(int row) { myRow = row; }
     public void run() {
       while (!done()) {
         processRow(myRow);

try {
           barrier.await();
         } catch (InterruptedException ex) {
return;
         } catch (BrokenBarrierException ex) {
return;
         }
       }
     }
   }

public Solver(float[][] matrix) {
     data = matrix;
     N = matrix.length;
     barrier = new CyclicBarrier(N,
                                 new Runnable() {
                                   public void run() {
                                     mergeRows();
                                   }
                                 });
     for (int i = 0; i < N; ++i)
       new Thread(new Worker(i)).start();

waitUntilDone();
}
}

在这个例子中，每个 worker 线程处理矩阵的一行，在处理完所有的行之前，该线程将一直在屏障处等待。
处理完所有的行之后，将执行所提供的 Runnable 屏障操作，并合并这些行。
如果合并者确定已经找到了一个解决方案，那么 done() 将返回 true，所有的 worker 线程都将终止。
如果屏障操作在执行时不依赖于正挂起的线程，则线程组中的任何线程在获得释放时都能执行该操作。
为方便此操作，每次调用 await() 都将返回能到达屏障处的线程的索引。然后，您可以选择哪个线程应该执行屏障操作，例如：
if (barrier.await() == 0) {
// log the completion of this iteration
}对于失败的同步尝试，CyclicBarrier 使用了一种要么全部要么全不 (all-or-none) 的破坏模式：如果因为中断、失败或者超时等原因，导致线程过早地离开了屏障点，那么在该屏障点等待的其他所有线程也将通过 BrokenBarrierException（如果它们几乎同时被中断，则用 InterruptedException）以反常的方式离开。

内存一致性效果：线程中调用 await() 之前的操作 happen-before 那些是屏障操作的一部份的操作，后者依次 happen-before 紧跟在从另一个线程中对应 await() 成功返回的操作。

(1)await

public int await() throws InterruptedException, BrokenBarrierException在所有参与者都已经在此 barrier 上调用 await 方法之前，将一直等待。
如果当前线程不是将到达的最后一个线程，出于调度目的，将禁用它，且在发生以下情况之一前，该线程将一直处于休眠状态：
最后一个线程到达；或者
其他某个线程中断当前线程；或者
其他某个线程中断另一个等待线程；或者
其他某个线程在等待 barrier 时超时；或者
其他某个线程在此 barrier 上调用 reset()。
如果当前线程：

在进入此方法时已经设置了该线程的中断状态；或者
在等待时被中断
则抛出 InterruptedException，并且清除当前线程的已中断状态。
如果在线程处于等待状态时 barrier 被 reset()，或者在调用 await 时 barrier 被损坏，抑或任意一个线程正处于等待状态，则抛出 BrokenBarrierException 异常。

如果任何线程在等待时被中断，则其他所有等待线程都将抛出 BrokenBarrierException 异常，并将 barrier 置于损坏状态。

如果当前线程是最后一个将要到达的线程，并且构造方法中提供了一个非空的屏障操作，则在允许其他线程继续运行之前，当前线程将运行该操作。
如果在执行屏障操作过程中发生异常，则该异常将传播到当前线程中，并将 barrier 置于损坏状态。

返回：
到达的当前线程的索引，其中，索引 getParties() - 1 指示将到达的第一个线程，零指示最后一个到达的线程
抛出：
InterruptedException - 如果当前线程在等待时被中断
BrokenBarrierException - 如果另一个线程在当前线程等待时被中断或超时，或者重置了 barrier，或者在调用 await 时 barrier 被损坏，抑或由于异常而导致屏障操作（如果存在）失败。

(2)getNumberWaiting

public int getNumberWaiting()返回当前在屏障处等待的参与者数目。此方法主要用于调试和断言。

返回：
当前阻塞在 await() 中的参与者数目。

应用实例：

package com.itm.thread;

import java.util.Random;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CyclicBarrierTest {
    public static void main(String[] args) {
        final CyclicBarrier cb = new CyclicBarrier(3); // 这个团队中共有3个队员,即需要3个线程
        ExecutorService es = Executors.newFixedThreadPool(3); // 在线程池中放入三个线程
        for (int i = 0; i < 3; i++) { // 开启三个任务
            es.execute(new Runnable() {

@Override
                public void run() {
                    for (int i = 0; i < 5; i++) {
                        try {
                            Thread.sleep(new Random().nextInt(5000));
                            System.out.print(Thread.currentThread().getName()
                                    + "已到达集合点" + (i + 1) + "，现在共有"
                                    + (cb.getNumberWaiting() + 1) + "个线程到达");
                            // 如果有2个线程已经在等待，那么最后一个线程到达后就可以一起开始后面操作
                            if (cb.getNumberWaiting() + 1 == 3) {
                                System.out.println("，全部到齐，出发去下一个目标");
                            } else {
                                System.out.println("，正在等待");
                            }
                            cb.await();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
            });
        }
        es.shutdown();
    }
}

运行结果：

pool-1-thread-3已到达集合点1，现在共有1个线程到达，正在等待
pool-1-thread-2已到达集合点1，现在共有2个线程到达，正在等待
pool-1-thread-1已到达集合点1，现在共有3个线程到达，全部到齐，出发去下一个目标
pool-1-thread-3已到达集合点2，现在共有1个线程到达，正在等待
pool-1-thread-1已到达集合点2，现在共有2个线程到达，正在等待
pool-1-thread-2已到达集合点2，现在共有3个线程到达，全部到齐，出发去下一个目标
pool-1-thread-1已到达集合点3，现在共有1个线程到达，正在等待
pool-1-thread-2已到达集合点3，现在共有2个线程到达，正在等待
pool-1-thread-3已到达集合点3，现在共有3个线程到达，全部到齐，出发去下一个目标
pool-1-thread-1已到达集合点4，现在共有1个线程到达，正在等待
pool-1-thread-2已到达集合点4，现在共有2个线程到达，正在等待
pool-1-thread-3已到达集合点4，现在共有3个线程到达，全部到齐，出发去下一个目标
pool-1-thread-2已到达集合点5，现在共有1个线程到达，正在等待
pool-1-thread-3已到达集合点5，现在共有2个线程到达，正在等待
pool-1-thread-1已到达集合点5，现在共有3个线程到达，全部到齐，出发去下一个目标

CyclicBarrier 和CountDownLatch使用详解相关推荐

CountDownLatch用法详解
CountDownLatch用法详解 CountDownLatch使用场景线程计数器用于线程执行任务,计数等待线程结束用法一: 等待所有的事情都做完 //程序计数器CountDownLatch ...
Java—CountDownLatch使用详解
关注微信公众号:CodingTechWork,一起学习进步. CountDownLatch介绍 CountDownLatch概述 CountDownLatch一般用作多线程倒计时计数器,强制它们等待其 ...
CountDownLatch原理详解
介绍当你看到这篇文章的时候需要先了解AQS的原理,因为本文不涉及到AQS内部原理的讲解. CountDownLatch是一种同步辅助,让我们多个线程执行任务时,需要等待线程执行完成后,才能执行下面的 ...
Java并发工具CountDownLatch使用详解
本文目录 1.使用场景 2.使用介绍 3.使用案例 4. Thread.join()和CountDownLatch的区别 1.使用场景通过使用 CountDownLatch可以使当前线程阻塞,等待其 ...
Java并发编程的艺术笔记（七）——CountDownLatch、CyclicBarrier详解
一.等待多线程完成的CountDownLatch CountDownLatch允许一个或多个线程等待其他线程完成操作,像加强版的join.(t.join()是等待t线程完成) 例: (1)开启多个线程 ...
并发工具类使用详解及区别（CountDownLatch、CyclicBarrier、Semaphore、Exchanger）
本文转载自:码农历险记 CountDownLatch CountDownLatch介绍 CountDownLatch是一个同步工具类,它允许一个或多个线程一直等待,直到其他线程执行完后再执行.例如,应 ...
Java多线程系列(九)：CountDownLatch、Semaphore等4大并发工具类详解
之前谈过高并发编程系列:4种常用Java线程锁的特点,性能比较.使用场景 ,以及高并发编程系列:ConcurrentHashMap的实现原理(JDK1.7和JDK1.8) 今天主要介绍concurre ...
Java并发编程之CountDownLatch/CyclicBarrierDemo/SemaphoreDemo详解
CountDownLatch详解什么是CountDownLatch? 代码说明一 :班长锁门代码说明二:秦国统一六国什么是CyclicBarrierDemo? 代码说明一:集齐7个龙珠,召唤神龙 ...
Java并发编程之CyclicBarrier详解
简介栅栏类似于闭锁,它能阻塞一组线程直到某个事件的发生.栅栏与闭锁的关键区别在于,所有的线程必须同时到达栅栏位置,才能继续执行.闭锁用于等待事件,而栅栏用于等待其他线程. CyclicBarrier ...

CyclicBarrier 和CountDownLatch使用详解

CyclicBarrier 和CountDownLatch使用详解相关推荐

最新文章

热门文章