Java并发框架Executor学习——CountDownLatch和CyclicBarrier

Java SE5的java.util.concurrent引入了大量设计用来解决并发问题的新类，这些类可以帮助程序员编写更加简单和健壮的并发程序

 

CountDownLatch

它被用来同步一个或多个任务，强制它们等待由其他任务执行的一组操作完成。其实可以把它看作一个计数器，只不过这个计数器的操作是原子操作，同时只能有一个线程去操作这个计数器，也就是同时只能有一个线程去减这个计数器里面的值。 

CountDownLatch的一个非常典型的应用场景是：有一个任务想要往下执行，但必须要等到其他的任务执行完毕后才可以继续往下执行。假如我们这个想要继续往下执行的任务调用一个CountDownLatch对象的await()方法，其他的任务执行完自己的任务后调用同一个CountDownLatch对象上的countDown()方法，这个调用await()方法的任务将一直阻塞等待，直到这个CountDownLatch对象的计数值减到0为止。 

CountDownLatch可以应用的许多实例，如文件的断点续传，将文件切割后分别下载，然后各个部分下载完成后再合并文件。

 

主要方法：

 

public void countDown()

对CountDownLatch的count做减一操作，如果count减到0，就释放所有等待的线程

 

public void await() throws InterruptedException

使当前线程阻塞，直到count减为0

 

public boolean await(long timeout, TimeUnit unit) throws InterruptedException

使当前线程阻塞，直到count减为0或阻塞时间超过timeout

 

举个例子，有三个工人在为老板干活，这个老板有一个习惯，就是当三个工人把一天的活都干完了的时候，他就来检查所有工人所干的活。记住这个条件：三个工人先全部干完活，老板才检查。所以在这里用Java代码设计两个类，Worker代表工人，Boss代表老板，具体的代码实现如下：

 

public class CountDownLatchDemo {
	public class Worker implements Runnable{
		private CountDownLatch latch;
		private String name;
		public Worker(String name,CountDownLatch latch) {
			this.name = name;
			this.latch = latch;
		}
		
		@Override
		public void run() {
			doWork();
			try {
				TimeUnit.SECONDS.sleep(new Random().nextInt(5)); 
				//完成任务，使count减1
				latch.countDown();
			} catch (Exception e) {
				e.printStackTrace();
			}
			System.out.println(name+": 我的工作完成了");
		}
		public void doWork(){
			System.out.println(name+"：正在工作");
		}
		
	}
	
	public class Boss implements Runnable{
		private CountDownLatch latch;
		Boss(CountDownLatch latch){
			this.latch = latch;
		}
		@Override
		public void run() {
			try {
				//一直阻塞，直到Worker完成工作
				latch.await();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			System.out.println("Boss检查工作");
		}
		
	}
	
	public void test(){
		CountDownLatch latch = new CountDownLatch(3);
		ExecutorService executor = Executors.newFixedThreadPool(4);
		
		Worker worker1 = new Worker("张三",latch);
		Worker worker2 = new Worker("李斯",latch);
		Worker worker3 = new Worker("王五",latch);
		Boss boss = new Boss(latch);
		executor.execute(worker1);
		executor.execute(worker2);
		executor.execute(worker3);
		executor.execute(boss);
		executor.shutdown();
	}
	
	public static void main(String[] args) {
		new CountDownLatchDemo().test();
	}
}

 

CyclicBarrier

CyclicBarrier适用于这样的情况：你希望创建一组任务，它们并行地执行工作，然后在进行下一个步骤之前等待，直至所有任务都完成。它使得所有的并行任务都将在栅栏处(Barrier)处列队，因此可以一致地向前移动。与CountDownLatch非常相似，只是CountDownLatch只触发一次事件，而CyclicBarrier可以多次重用。因此从字面意思可以翻译为“循环的栅栏”。

 

主要方法：

 

public int await() throws InterruptedException, BrokenBarrierException

使当前线程阻塞，直到所有的并行任务都执行了一轮

 

public int await(long timeout, TimeUnit unit)throws InterruptedException,BrokenBarrierException,TimeoutException 

使当前线程阻塞，直到所有的并行任务都执行了一轮或阻塞时间超过timeout

 

public int getNumberWaiting() 

获取正在等待栅栏动作的任务数

 

应用实例：赛马游戏。本实例中有7匹马一起跑，每匹马每跑一轮，统计一下这几匹马的比赛情况，利用打印"*"的长度表示马跑的路程。本游戏利用CyclicBarrier实现，每一匹马是一个线程任务，它们并行执行，统计赛马情况是一个栅栏动作(barrierAction)。具体代码实现如下：

 

public class CyclicBarrierDemo {
	class Horse implements Runnable{
		private int strides = 0;
		private Random random = new Random(47);
		private int id;
		private CyclicBarrier barrier;
		public Horse(int id,CyclicBarrier barrier){
			this.id = id;
			this.barrier = barrier;
		}
		
		@Override
		public void run() {
			try {
				while(!Thread.interrupted()){
					synchronized (this) {
						strides += random.nextInt(3);
					}
				}
				//通知CyclicBarrier，一个任务已经完成
				barrier.await();
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (BrokenBarrierException e) {
				e.printStackTrace();
			}
		}
		
		//返回马跑的路程
		public String track(){
			StringBuilder sb = new StringBuilder();
			for(int i=0;i<strides;i++){
				sb.append("*");
			}
			sb.append(id);
			return sb.toString();
		}
		
		public synchronized int getStrides(){
			return strides;
		}

		@Override
		public String toString() {
			return "Horse[id=" + id + "] ";
		}
	}
	
	class HorseRace{
		private final int FINISH_LINE = 100;
		private CyclicBarrier barrier;
		private List<Horse> horses = new ArrayList<Horse>();
		private ExecutorService executor;
		public HorseRace(int nHorse){
			//7匹马没跑一轮，栅栏动作就触发
			barrier = new CyclicBarrier(nHorse, new Runnable() {
				
				@Override
				public void run() {
					StringBuilder sb = new StringBuilder();
					for(int i=0;i<FINISH_LINE;i++){
						sb.append("=");
					}
					System.out.println(sb.toString());
					
					//输出每匹马的赛况
					for(Horse h : horses){
						System.out.println(h.track());
					}
					
					//有一匹马跑到终点，游戏结束
					for(Horse h : horses){
						if(h.getStrides() >= FINISH_LINE){
							System.out.println(h+" win!");
							executor.shutdownNow();
							return;
						}
					}
					try {
						TimeUnit.MILLISECONDS.sleep(200); 
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
			});
			executor = Executors.newCachedThreadPool();
			for(int i=0;i<nHorse;i++){
				Horse horse = new Horse(i+1,barrier);
				horses.add(horse);
				executor.execute(horse);
			}
		}
	}
	
	public void test(){
		new HorseRace(7);
	}
	
	public static void main(String[] args) {
		new CyclicBarrierDemo().test();
	}
}