Java多线程：CountDownLatch、CyclicBarrier 和 Semaphore

场景描述：

　　多线程设计过程当中，常常会遇到须要等待其它线程结束之后再作其余事情的状况。

有几种方案：

 

　　1.在主线程中设置一自定义全局计数标志，在工做线程完成时，计数减1。主线程侦测该标志是否为0，一旦为0，表示全部工做线程已经完成。

　　2.使用Java标准的类CountDownLatch来完成这项工做，原理是同样的，计数。

 

 

CountDownLatch

一个同步辅助类，在完成一组正在其余线程中执行的操做以前，它容许一个或多个线程一直等待。 

其机制是：

　　当多个（具体数量等于初始化CountDownLatch时的count参数的值）线程都达到了预期状态或完成预期工做时触发事件，其余线程能够等待这个事件来触发本身的后续工做。这里须要注意的是，等待的线程能够是多个，即CountDownLatch是能够唤醒多个等待的线程的。达到本身预期状态的线程会调用CountDownLatch的countDown方法，而等待的线程会调用CountDownLatch的await方法。

CountDownLatch 很适合用来将一个任务分为n个独立的部分，等这些部分都完成后继续接下来的任务，CountDownLatch 只能出发一次，计数值不能被重置。

流程图

 

如上图所示，当7个线程都完成latch.countDown调用后，最下面那条线程会从latch.await返回，继续执行后面的代码

函数列表

• CountDownLatch(int count) ：构造一个用给定计数初始化的 CountDownLatch。

• void await()：使当前线程在锁存器倒计数至零以前一直等待，除非线程被中断。

• boolean await(long timeout, TimeUnit unit) 使当前线程在锁存器倒计数至零以前一直等待，除非线程被中断或超出了指定的等待时间。

• void countDown() 递减锁存器的计数，若是计数到达零，则释放全部等待的线程。

实现原理

　　请参考：http://www.tuicool.com/articles/mQnAfq

 

实例

 　　咱们来看一个具体的例子。假设咱们使用一台多核的机器对一组数据进行排序，那么咱们能够把这组数据分到不一样线程中进行排序，而后合并；能够利用线程池来管理多线程；能够将CountDownLatch用做各个分组数据都排好序的通知。下面是代码片断：

先看主线程

int count = 10;
final CountDownLatch latch = new CountDownLatch(count);
int[] datas = new int[10204];
int step = datas.length / count;
for (int i=0; i < count; i++) {
    int start = i * step;
    int end = (i+1) * step;
    if (i == count - 1) end = datas.length;
    threadpool.execute(new MyRunnable(latch, datas, start, end));
}
latch.await();
//合并数据

咱们再看一下具体任务的代码，即MyRunnable的run方法的实现：

public void run() {
      //数据排序
     latch.countDown(); 
}

 

CyclicBarrier

能够协同多个线程，让多个线程在这个屏障前等待，直到全部线程都达到了这个屏障时，再一块儿继续执行后面的动做。

CyclicBarrier适用于多个线程有固定的多步须要执行，线程间互相等待，当都执行完了，再一块儿执行下一步。由于该 barrier 在释放等待线程后能够重用，因此称它为循环 的 barrier。

 

流程图

 上图中的7个线程各有一个barrier.await,那么任何一个线程在执行到barrier.await时就会进入阻塞等待状态，直到7个线程都到了barrier.await时才会同时从await返回，继续后面的工做。此外若是在构造CyclicBarrier时设置了一个Runnable实现，那么最后一个到barrier.await的线程会执行这个Runnable的run方法，以完成一些预设的工做。

 

注意比较CountDownLatch和CyclicBarrier：

　　(01) CountDownLatch的做用是容许1或N个线程等待其余线程完成执行；而CyclicBarrier则是容许N个线程相互等待。

　　(02) CountDownLatch的计数器没法被重置；CyclicBarrier的计数器能够被重置后使用，所以它被称为是循环的barrier。

CountDownLatch 适用于一组线程和另外一个主线程之间的工做协做。一个主线程等待一组工做线程的任务完毕才继续它的执行是使用 CountDownLatch 的主要场景；CyclicBarrier 用于一组或几组线程，好比一组线程须要在一个时间点上达成一致，例如同时开始一个工做。另外，CyclicBarrier 的循环特性和构造函数所接受的 Runnable 参数也是 CountDownLatch 所不具有的。

 

 

	CountDownLatch	CyclicBarrier
适用场景	主线程等待其余工做线程结束	多个线程相互等待，直到全部线程都达到一个障碍点Barrier
主要方法	CountDownLatch(int count) 主线程调用：初始化计数   await() 主线程调用 ： 阻塞，直到等待计数为0时解除阻塞    countDown() 工做线程调用 ： 计数减1	CyclicBarrier(int parties , Runnnable barrierAction) : 初始化参与者数量和障碍点执行Action，action可选，由主线程初始化   await() : 由工做线程调用，每被调用一次，计数便会减小1，并阻塞住当前线程 ， 直到全部线程都达到障碍点
等待结束	各线程之间再也不相互影响， 能够继续作本身的事情， 再也不执行下一个工做目标。	在障碍点到达后， 容许全部线程继续执行，到达下一个目标后，能够恢复使用CyclicBarrier， barrier 在释放等待线程后能够重用
异常		若是其中一个线程因为中断、错误、或者超时致使永久离开障碍点，其余线程也将抛出异常。

 

实例

 　　

int count = 10;
final CyclicBarrier barrier = new CyclicBarrier(count + 1);
int[] datas = new int[10204];
int step = datas.length / count;
for (int i=0; i < count; i++) {
    int start = i * step;
    int end = (i+1) * step;
    if (i == count - 1) end = datas.length;
    threadpool.execute(new MyRunnable(barrier, datas, start, end));
}
barrier.await();
//合并数据

能够看到CyclicBarrier对象传入的参数值比CountDownLatch大1，缘由是构造CountDownLatch的参数是调用countDown的数量，而CyclicBarrier的数量是await的数量

public void run() {
      //数据排序
     try {
         barrier.await(); 
    }catch (...)
}

 

Semaphore

Semaphore 信号量对象管理的信号就像令牌，构造时传入个数，总数就是控制并发的数量 。咱们须要控制并发的代码，执行前先获取信号(经过acquire获取信号许可)，执行后归还信号(经过release归还信号许可)。每次acquire成功返回后，Semaphore可用的信号量就会减小一个，若是没有可用的信号，acquire调用就会阻塞，等待有release调用释放信号后，acquire才会获得信号并返回。

若是Semaphore管理的信号量为1个，那么就退化到互斥锁了；若是多于一个信号量，则主要用于控制并发数。与经过控制线程数来控制并发数的方式相比，经过Semaphore来控制并发数能够控制得更加细粒度，由于真正被控制最大并发的代码放到acquire和release之间就好了。

　　Semaphore类位于java.util.concurrent包下，它提供了2个构造器：

public Semaphore(int permits) {          //参数permits表示许可数目，即同时能够容许多少线程进行访问
    sync = new NonfairSync(permits);
}
public Semaphore(int permits, boolean fair) {    //这个多了一个参数fair表示是不是公平的，即等待时间越久的越先获取许可
    sync = (fair)? new FairSync(permits) : new NonfairSync(permits);
}

实例

 　　例如咱们须要控制远程方法的并发量，超过并发量的方法就等待有其余方法执行返回后再执行，那么其代码以下：

semaphore.acquire();
try {
    //调用远程通讯的方法
}
finally {
    semaphore.release();
}

 

 

 

 

参考资料：

    Java并发编程：CountDownLatch、CyclicBarrier和Semaphore