Java多线程之CountDownLatch、CyclicBarrier和Semaphore

转自：http://www.liubey.org/countdownlatch_vs_cyclicbarrier/

概述

CountDownLatch : 一个线程(或者多个)， 等待另外N个线程完成某个事情之后才能执行。
CyclicBarrier : N个线程相互等待，任何一个线程完成之前，所有的线程都必须等待。
Semaphore：可以控制某个资源可被同时访问的个数，通过 acquire() 获取一个许可，如果没有就等待，而 release() 释放一个许可。

举例

1：CountDownLatch

使用场景举例：

体育课时老师拿着秒表测试同学的800米成绩，那需求就是很简单了，老师在起跑处组织大家一起跑的瞬间按下秒表计时开始，然后再终点处等待最后一个学生通过终点后开始汇集学生成绩。

API相关：

1）await(),阻塞等待，直到计数器清零
2）await(int timeout, TimeUnit unit),使线程阻塞，除非被中断或者超过等待的最大时间
如果达到计数器清零，则await返回true，如果等待超过了最大的等待时间，则返回false
3）countDown(),计数器减一，当计数器清零时，await的线程被唤醒，线程继续执行
4）getCount ()，获取当前计数器的大小

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class TeacherWithStopwatch {
public static final int NUMBER_OF_STUDENT = 10;
public static final int NUMBER_OF_TEACHER = 1;
public static void main(String[] args) {
ExecutorService executor = Executors.newFixedThreadPool(2);
CountDownLatch studentSignal = new CountDownLatch(NUMBER_OF_STUDENT);
CountDownLatch teacherSignal = new CountDownLatch(NUMBER_OF_TEACHER);
for (int i = 0; i < NUMBER_OF_STUDENT; i++) {
executor.execute(new Student(i, studentSignal, teacherSignal));
}
try {
System.out.println("各就各位！开跑！");
teacherSignal.countDown();
studentSignal.await();
System.out.println("结果发送到汇报成绩的系统");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
executor.shutdown();
}
}
}

import java.util.concurrent.CountDownLatch;
public class Student implements Runnable {
private int id;
private CountDownLatch studentSignal;
private CountDownLatch teacherSignal;
public Student(int id, CountDownLatch studentSignal,
CountDownLatch teacherSignal) {
this.id = id;
this.studentSignal = studentSignal;
this.teacherSignal = teacherSignal;
}
@Override
public void run() {
try {
teacherSignal.await();
System.out.println("学生" + id + "起跑...");
System.out.println("学生" + id + "到达终点。");
studentSignal.countDown();
System.out.println("学生" + id + "继续干其他事情");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

2：CyclicBarrier

使用场景举例：

有四个游戏玩家玩游戏，游戏有三个关卡，每个关卡必须要所有玩家都到达后才能允许通关。
其实这个场景里的玩家中如果有玩家A先到了关卡1，他必须等待其他所有玩家都到达关卡1时才能通过。
也就是说线程之间需要互相等待，这和CountDownLatch的应用场景有区别，
CountDownLatch里的线程是到了运行的目标后继续干自己的其他事情，而这里的线程需要等待其他线程后才能继续完成下面的工作。

API相关：

public CyclicBarrier(int parties) 创建一个新的 CyclicBarrier，
它将在给定数量的参与者（线程）处于等待状态时启动，但它不会在启动 barrier 时执行预定义的操作。
public CyclicBarrier(int parties, Runnable barrierAction) 创建一个新的 CyclicBarrier，
它将在给定数量的参与者（线程）处于等待状态时启动，并在启动 barrier 时执行给定的屏障操作，
该操作由最后一个进入 barrier 的线程执行。
public int await() throws InterruptedException, BrokenBarrierException 在所有参与者都已经在此 barrier 上调用 await 方法之前，将一直等待。
public int await(long timeout,TimeUnit unit) throws InterruptedException, BrokenBarrierException,TimeoutException
在所有参与者都已经在此屏障上调用 await 方法之前将一直等待,或者超出了指定的等待时间。
public int getNumberWaiting() 返回当前在屏障处等待的参与者数目。此方法主要用于调试和断言。
public int getParties() 返回要求启动此 barrier 的参与者数目。
public boolean isBroken() 查询此屏障是否处于损坏状态。
public void reset() 将屏障重置为其初始状态。

public class GameBarrier {
public static final int NUMBER_OF_PLAYERS = 4;
public static void main(String[] args) {
ExecutorService executor = Executors.newFixedThreadPool(NUMBER_OF_PLAYERS);
CyclicBarrier barrier = new CyclicBarrier(NUMBER_OF_PLAYERS, new Runnable() {
@Override
public void run() {
System.out.println("所有玩家通过第一关！");
}
});
for (int i = 0; i < NUMBER_OF_PLAYERS; i++) {
executor.execute(new Player(i, barrier));
}
executor.shutdown();
}
}

public class Player implements Runnable {
private CyclicBarrier cyclicBarrier;
private int id;
public Player(int id, CyclicBarrier cyclicBarrier) {
this.cyclicBarrier = cyclicBarrier;
this.id = id;
}
@Override
public void run() {
try {
System.out.println("玩家" + id + "通过第一关...");
cyclicBarrier.await();
System.out.println("玩家" + id + "进入第二关...");
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}
}

3：Semaphore

使用场景举例：

一个资源池只能同时五个人使用，那么十个人来轮询使用的情况就是每个人都先申请资源，使用完归还于下一个人使用。

API相关：

构造器中的fairness为true时，Semaphore保证各线程以后进先出（FIFO）的方式获得信号量。如果fairness为false，则不保证这种顺序，允许各线程之间的“讨价还价”。
tryAcquire与release为主要方法

public class Person implements Runnable {
private SemaphorePool pool;
private int id;
public Person(int id, SemaphorePool pool) {
this.pool = pool;
this.id = id;
}
@Override
public void run() {
try {
pool.applyResource();
System.out.println("人物" + id + "进入");
Thread.sleep(1000);
System.out.println("人物" + id + "离开");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
pool.releaseResource();
}
}
}

public class SemaphorePool {
private Semaphore canExecuteCount = new Semaphore(5, true);
private static final int TRY_EXECUTE_TIMEOUT = 20;
public boolean applyResource() {
boolean canExecute = false;
try {
canExecute = canExecuteCount.tryAcquire(1, TRY_EXECUTE_TIMEOUT,
TimeUnit.SECONDS);
} catch (InterruptedException e) {}
return canExecute;
}
public void releaseResource() {
canExecuteCount.release(1);
}
}

public class TestSemaphore {
public static final int NUMBER_OF_PERSONS = 10;
public static final SemaphorePool pool = new SemaphorePool();
public static void main(String[] args) {
ExecutorService executor = Executors.newFixedThreadPool(NUMBER_OF_PERSONS);
for(int i=0;i<NUMBER_OF_PERSONS;i++) {
executor.execute(new Person(i, pool));
}
executor.shutdown();
}
}