多线程之并发流程控制方式

多线程间经常需要协调一批任务线程同时完成之后去做某事，而Java中提供的流程控制有两种方式：一种是CyclicBarrier，另一种是CountDownLatch.

 

第一种方式 CyclicBarrier。来自jdk的解释：一个同步辅助类，它允许一组线程互相等待，直到到达某个公共屏障点 (common barrier point)。在涉及一组固定大小的线程的程序中，这些线程必须不时地互相等待，此时 CyclicBarrier 很有用。因为该 barrier 在释放等待线程后可以重用，所以称它为循环 的 barrier。 CyclicBarrier 支持一个可选的 Runnable 命令，在一组线程中的最后一个线程到达之后（但在释放所有线程之前），该命令只在每个屏障点运行一次。若在继续所有参与线程之前更新共享状态，此屏障操作 很有用。

 

第二种方式 CountDownLatch。来自jdk的解释：一个同步辅助类，在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待。 用给定的计数 初始化 CountDownLatch。由于调用了 countDown() 方法，所以在当前计数到达零之前，await 方法会一直受阻塞。之后，会释放所有等待的线程，await 的所有后续调用都将立即返回。这种现象只出现一次——计数无法被重置。如果需要重置计数，请考虑使用 CyclicBarrier。 CountDownLatch 是一个通用同步工具，它有很多用途。将计数 1 初始化的 CountDownLatch 用作一个简单的开/关锁存器，或入口：在通过调用 countDown() 的线程打开入口前，所有调用 await 的线程都一直在入口处等待。用 N 初始化的 CountDownLatch 可以使一个线程在 N 个线程完成某项操作之前一直等待，或者使其在某项操作完成 N 次之前一直等待。 CountDownLatch 的一个有用特性是，它不要求调用 countDown 方法的线程等到计数到达零时才继续，而在所有线程都能通过之前，它只是阻止任何线程继续通过一个 await。

 

 源码示例如下：

 

package multiThreadShow;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;

public class MultiThreadShow {

public static void main(String[] args){
		
		HashMap<Integer, ArrayList<Integer>>  map = new HashMap<Integer,ArrayList<Integer>>(); 
	 
		HashMap<Integer, ArrayList<Integer>>  map2 = new HashMap<Integer,ArrayList<Integer>>();
		for(int i=0;i<50;i++){
			ArrayList<Integer> arrayList= new ArrayList<Integer>(); 
			for(int i1=0;i1<10000;i1++){
				int p = (int) (Math.random()*10000);
				arrayList.add(p);
			}
			map.put(i, arrayList);
		}
		
		for(int i=0;i<50;i++){
			ArrayList<Integer> arrayList= new ArrayList<Integer>(); 
			for(int i1=0;i1<10000;i1++){
				int p = (int) (Math.random()*10000);
				arrayList.add(p);
			}
			map2.put(i, arrayList);
		}
		CountDownLatch doneSignal = new CountDownLatch(map.size());

		long start = System.currentTimeMillis();
		for(int i : map.keySet()){
			 SortThread t= new SortThread(doneSignal,"SortThread-"+i);
			 t.setArrayList(map.get(i));
			 t.start();
		}
		
		try {
			doneSignal.await();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		long end = System.currentTimeMillis();	
		
		System.out.println("Excuting this totally costs "+(end - start));
		
		start = System.currentTimeMillis();
		final long input = start;
		
		//其中的Runnable任务用来统计这批任务的执行时间   当所有的任务都达到cyclicBarrier点时（即表示这些任务都完成）
		//那么通过这个Runnable任务去统计执行时间
		CyclicBarrier cyclicBarrier = new CyclicBarrier(map2.size(),new Runnable(){

			@Override
			public void run() {
				// TODO Auto-generated method stub
				System.out.println("Excuting this totally costs "+(System.currentTimeMillis() - input));
			}
			
		});
		
		
		for(int i : map2.keySet()){
			SortThreadCyclicBarrier t= new SortThreadCyclicBarrier(cyclicBarrier,"SortThreadCyclicBarrier-"+i);
			 t.setArrayList(map2.get(i));
			 t.start();
		}
		
		// end = System.currentTimeMillis();	
		//main 这个线程执行的顺序可能在阻塞期间 所以这个时间点没有用
		//System.out.println("Excuting this totally costs "+(end - start));
		
	}
}
//CountDownLatch排序Thread
class SortThread extends Thread{
	private  ArrayList<Integer> arrayList;
	private final CountDownLatch doneSignal;

	public ArrayList<Integer> getArrayList() {
		return arrayList;
	}

	public void setArrayList(ArrayList<Integer> arrayList) {
		this.arrayList = arrayList;
	}

	public SortThread(CountDownLatch doneSignal,String name){
		super(name);
		this.doneSignal = doneSignal;
	}
	
	public void run(){
		Collections.sort(arrayList);
		doneSignal.countDown();
		//System.out.println("Sort finished"+this.getName());
	}
	
}
//CyclicBarrier排序Thread
class SortThreadCyclicBarrier extends Thread{
	private  ArrayList<Integer> arrayList;
	private final CyclicBarrier cyclicBarrier;

	public ArrayList<Integer> getArrayList() {
		return arrayList;
	}

	public void setArrayList(ArrayList<Integer> arrayList) {
		this.arrayList = arrayList;
	}

	public SortThreadCyclicBarrier(CyclicBarrier cyclicBarrier,String name){
		super(name);
		this.cyclicBarrier = cyclicBarrier;
	}
	
	public void run(){
		System.out.println("Sort Start "+this.getName());
		Collections.sort(arrayList);
		System.out.println("Sort finished "+this.getName());
		try {
			cyclicBarrier.await();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			return;
		} catch (BrokenBarrierException e) {
			// TODO Auto-generated catch block
			return;
		}
		//System.out.println("Sort finished "+this.getName());
		
	}
	
} 

 

 

 说明：这两种流程控制手段的区别在于 后者可以循环使用，而且后者控制的线程一个出现问题，则剩下的线程都会以抛出BrokenBarrierException 的方式终止。

评论

1 楼 SINCE1978 2011-02-04  

作为示例程序过于复杂了