Java并发编程（三） CyclicBarrier

在上两回中，用CountDownLatch可以很好的让4个工人等待轮胎运送到工作间合适的位置后再进行轮胎装配工作，装配完4个轮子后，4个工人用减少CountDownLatch计算器数值的方式告诉生产线，这辆汽车的轮子已经装配好。可是，你有没有发现，在我们的程序中，这4个工人只装配了一辆汽车，假如这个4个工人一天只装配一辆汽车，而你是老板的话，你可以解雇这4个工人，工厂不需要生产效率这样低下的劳动者。现在，工厂需要4个工人一天必须装配完100辆汽车的轮胎。4个工人说，这不是我们的问题，流水线上只有一辆汽车可以装配。那么，作为管理方，有办法重新设计流水线的作业方式，让4个工人舒舒服服的在一天之内安装100辆车的轮胎吗？答案是肯定的，我们有办法！于是我们重新设计流水线的作业方式。

 

流水线作业新方式

1. 当工人们就位后，流水线上已经有一辆车放在他们的作业区，轮胎也被放置到作业区合适的位置，让工人很舒服的安装
2. 当4个工人安装好这辆车的4个轮胎后，工人们可以进入短暂的等待状态，等待下一辆汽车被放到作业区
3. 下一辆等待装配轮胎的汽车进入作业区。
4. 四个新的轮胎进入作业区内放置轮胎的位置。
5. 步骤3和步骤4准备好后，工人们可以进行装配轮胎的工作。
6. 4个工人和流水线配合，在一天内，重复3，4和5这三个步骤，直到100辆汽车的轮胎安装完成。
   

 

CyclicBarrier

通过观察流水线新的作业方式，我们知道，使用CountDownLatch已经不能适应新作业方式的需求。不过，CyclicBarrier类很有信心保证能完成控制流水线的工作。它可以让每个受它监视的线程只要进入等待状态后，自身启动一个动作，用于执行一些特定任务，这个动作执行完后，线程又进入运行状态。如此循环，直到整个工作完成。

CyclicBarrier有两个构造函数。

1. CyclicBarrier(int) 用于构造一个CyclicBarrier实例，参数int表示在这个CyclicBarrier上等待的线程的数量。
2. CyclicBarrier(int, Runnable) 用于构造一个带Runnable接口的实例，第一个int参数与第一种构造函数中的参数意义相同，runnable接口代表一个要执行的动作，这个动作在受到CyclicBarrier实例监视的所有线程都处于等待状态时执行。如果用第一种构造函数，则这些线程处于等待状态时，什么动作也不执行。

FullTimeWorker

现在这4个工人不再是上一回中，1天只能装配1个轮胎的工人了，他们有能力在一天之内装配100个轮胎，因此，我们称它们为全职工人，在FullTimeWorker线程中，当前的WorkSpace中的车辆已经放置好，工人拿到一个轮胎，安装上后，处于等待下一辆车和轮胎被放到工作区。FullTimeWorker线程中，如果捕捉到BrokenBarrierException，表示当前的流水线控制系统有问题，则会停止当前的工作，另外3个FullTimeWorker线程也会停止工作，整个装配工作将停止。

 

@Override
	public void run() {
		for (int i = 0; i < 100; i++) {
			try {
				Tyre tyre = tyries.take();  //拿到一个轮胎
				Car car = workspace.getCar();   //当前作业区中的车辆

//我开始进行装配工作
 System.out.format("%s is assembling the %s for the %s.%n", name, tyre, car);
				marshalTyre();
				System.out.format("%s has assembled the %s for the %s.%n", name, tyre, car);

				// 我的轮胎装配完了，等待下一辆车和轮胎被送入我的作业区。				
				barrier.await();

			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (BrokenBarrierException e) {
				e.printStackTrace();
				return;
			}
		}
		System.out.format("%s has done work of the day%n", name);
	}

 

AssemblyLine

AssemblyLine类是这个流水线上的控制程序，当CyclicBarrier锁发现这4个工人都处于等待状态时，控制流水线把下一辆车和4个新的轮胎送入这4个工人的作业区。让这个4个工作有事可做，而不是一直处于等待状态。

AssemblyLine必须实现Runnable接口。显然，运送轮胎和车的工作必须放在run方法中实现。

static class AssemblyLine implements Runnable {
		static int tyreNo = 0;
		ArrayBlockingQueue<Car> cars;

		@Override
		public void run() {
			try {
				Car car = null;
				if (!cars.isEmpty()) {
					car = cars.take();
					workspace.setCar(car);  // 把下一辆车放入作业区
					prepareTyre();                // 把4个轮胎放入作业区
					System.out.format("-------------- a new %s and four tyres are ready --------------%n", car);
				}
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}

		public AssemblyLine(ArrayBlockingQueue<Car> cars) {
			this.cars = cars;
		}

		private static void prepareTyre() throws InterruptedException {
			for (int i = 0; i < 4; i++) {
				tyries.offer(new Tyre(tyreNo++));
				Thread.sleep(500);
			}
		}
	}

 

运行

到目前为止，流水线作业方式已经重新设计过，工人也有能力一天安装100个轮胎，流水线运送也有能力自动的根据工人的状态来运送车辆和轮胎，所有的一切都已经具备，就等按下总开关，让整条流水线开始运作起来。

首先，准备好100辆车可供流水线可调配。

ArrayBlockingQueue<Car> cars = new ArrayBlockingQueue<Car>(100);
		for (int i = 0; i < 100; i++)
			cars.offer(new Car(i));

 

接着把工人们放到流水线位置，让CyclicBarrier作为流水线控制系统，协调工人与流水线的工作，这里有一点小小的注意，在工人们开始工作之前，流水线先放一辆车和4个轮胎到工作区。

CyclicBarrier barrier = new CyclicBarrier(4, new AssemblyLine(cars));

		AssemblyLine.prepareTyre();
		workspace.setCar(cars.take());

 

接着，工人们开始干活了，流水线也可以运转了。

ExecutorService executorService = Executors.newFixedThreadPool(4);
		String works[] = { "John Chembers", "Bill Gates", "Larry Page", "Ericsson Limited" };
		for (int i = 0; i < 4; i++)
			executorService.execute(new FullTimeWorker(works[i], barrier));

		executorService.shutdown();

 

小结

在一个CyclicBarrier监视下的所有线程都处于等待状态时，CyclicBarrier可以执行一个特定的动作。这个动作执行完后，线程继续运行。如果让最后一个进入等待的线程干一个特定的动作，可以在线程的run方法内进行判断，下面的代码假设CyclicBarrier实例共监视着4个线程

if (barrier.await() == 4) {
// some action...
}

 

如果任何一个被监视的线程在等待是发生了任何异常，所有其它的线程也会抛出同样的异常。在FullTimeWorker的run方法中，如果捕捉到了InterruptedException，其他的三个线程也会抛出InterruptedException。

 

最后，完整的代码在附近中，欢迎下载测试，并告诉我您所发现的bug.

 

[原创内容，版权所有，如有转载，请注明出处，如有错误之处，请指出，不胜感激]