Java分布式应用学习笔记05多线程下的并发同步器----后篇 博客分类： 分布式集群 分布式集群线程调度多线程同步器

5.  CountDownLatch

很多资料上都说CountDownLatch是倒数计数器，我觉得这种说法太过专业，其实它就是一个数数的人员。利用它，可以在多线程执行任务完毕后完成进行多线程的等待，便于等待所有的线程之后在干别的事情，这个有点类似于FutureTask，使用上不太一样。这个场景就是一个线程必须要等到其他线程执行完毕后才能往下执行，注意，这里这个线程没必要需要其他线程的执行结果，而是只有一个条件就是其他线程必须执行完毕。咱们依然做个比喻，韦小宝需要8部《四十二章经》才能去鹿鼎山找寻宝藏，怎么办，他有7个老婆，不能资源浪费啊，这个任务同时进行吧。咱们设计8个线程同时进行，等所有老婆都执行完毕了，每个老婆找齐了《四十二章经》了，好了，他可以自己去找宝藏了。等等，咱们小宝哥有7个老婆，何来8个线程，这个问题，读者不必较真，举个例子罢了，咱们给他加个老婆不就行了！

代码如下

package threadConcurrent.test;

import java.util.concurrent.CountDownLatch;

/**
 * 分部执行任务
 * 
 * @author liuyan
 * 
 */
public class CountDownLatchDemo implements Runnable {
	private int id;

	// 线程之间不影响，到了终点后继续做自己的事情
	private CountDownLatch countDownLatch;

	public CountDownLatchDemo(int id, CountDownLatch countDownLatch) {
		this.id = id;
		this.countDownLatch = countDownLatch;
	}

	@SuppressWarnings("static-access")
	@Override
	public void run() {
		try {

			System.out.println("第" + (id + 1) + "小老婆开始查找《四十二章经》...");
			Thread.currentThread().sleep(id * 1000);
			System.out.println("第" + (id + 1) + "本《四十二章经》找到");

			//计数器将等待数字-1
			countDownLatch.countDown();

			System.out.println("第" + (id + 1) + "小老婆继续干其他事情");
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	public static void main(String[] args) {
		CountDownLatch countDownLatch = new CountDownLatch(8);

		for (int i = 0; i < 8; i++) {
			new Thread(new CountDownLatchDemo(i, countDownLatch)).start();
		}

		try {

			System.out.println("韦小宝等着等着8本四十二章……");

			// 韦小宝等着等着
			countDownLatch.await();

			// 等待运动员到达终点
			System.out.println("8本四十二章经找寻完成，可以寻宝了！");
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
}

 

主线程就当做是韦小宝吧，主线程首先开辟了线程计数器对象，之后就开辟了8个线程，派出了8个小老婆去办事，之后主线程调用countDownLatch.await()阻塞，在家里一直喝着小茶，听着小曲等着8个线程的执行完毕。咱们来看小老婆们，小老婆们就辛苦了，开始找寻经书，之后调用countDownLatch.countDown()方法通知线程计数器减去1，让等待的线程减去1。就好比说有个小老婆找到了《四十二章经》，用iphone发个彩信将经书的夹缝地图发给韦小宝，韦小宝收到了，恩，这个老婆能干，任务完成，我的等待目标减去1。等到等待线程为0的时候，小宝开始行动了，将手机里的地图通过游戏——拼图游戏，一拼凑，大事可成，自己寻宝去！这里大家也看到了CountDownLatch与FutureTask的区别。CountDownLatch侧重的是分线程的完成个数，每次完成一个分线程，等待数目减少一个，等待线程为0的时候，主线程的就不阻塞了，开始往下走。而分线程一旦调用countDownLatch.countDown()方法，就代表分线程任务搞定，主线程就不会因为你的其他事情而不能往下走，完成任务了，小老婆们也可以去旅旅游，休息休息！而FutureTask则是注重执行结果的，主线程需要它的确切结果。所以futureTask执行的call()有返回值。

6.  CyclicBarrier

CyclicBarrier相对于CountDownLatch来说，最大的不同是，分线程具体的执行过程受其他分线程的影响，必须每个分线程都执行完毕了，主线程才继续往下走，而分线程如果在所有分线程执行完毕后还有其他动作，ok，还你*，不必阻塞了，往下走你的路吧。这个例子是网上的游戏玩家的例子，4个小孩玩游戏，游戏要求必须是4个小孩都得通过第一关，才能开启第二关的关口！否则其他完成第一关的人都得等着其他人完成。这个有点像我们的项目开发，分模块开发，到一定阶段将模块汇总，联调，测试，如果这时候有一个模块没完成，大家等着吧，大家都在那里静静地、盯着你、等着你。

package threadConcurrent.test;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class GameBarrier {
	public static void main(String[] args) {
		CyclicBarrier cyclicBarrier = new CyclicBarrier(4, new Runnable() {

			@Override
			public void run() {
				System.out.println("所有玩家进入第二关！");
			}
		});

		for (int i = 0; i < 4; i++) {
			new Thread(new Player(i, cyclicBarrier)).start();
		}
	}

}

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();
		}
	}
}

 使用cyclicBarrier.await();方法进行等待、阻塞，当所有分线程执行完毕了，主线程开始执行，分线程的*也解脱了，继续往下走，开始第二关。

7.  Exchanger

Exchanger是线程资源交换器，线程A与线程B在某个运行阶段需要互换资源才能完成任务。这就好比2个公司职员——叶小钗和一页书。分别在不同的项目组——组A和组B，两个组开发者不同的项目，在正常时候叶小钗在组A上班开发者BOSS项目，一页书在项目组B开发ESB中间件产品。而在特殊时期项目组B不需要一页书了，需要叶小钗提供技术支持，就和项目组A要叶小钗，项目组A的leader也不是吃素的，你要叶小钗，没问题，把一页书也得接我们项目组剥削几天！就这样项目组B与项目组A做了这种“交易”（交换），用完了之后，恩~看程序吧

package threadConcurrent.test;

import java.util.concurrent.Exchanger;

/**
 * 资源交换
 * @author liuyan
 */
public class ExgrDemo {
	public static void main(String args[]) {
		
		//交换器
		Exchanger<String> exgr = new Exchanger<String>();

		new TeamB(exgr);
		new TeamA(exgr);
	}
}

/**
 * 项目组A
 * @author liuyan
 */
class TeamA implements Runnable {
	Exchanger<String> ex;

	String str;

	TeamA(Exchanger<String> c) {
		ex = c;
		str = new String();

		new Thread(this).start();
	}

	public void run() {
		char ch = 'A';
		for (int i = 0; i < 3; i++) {
			for (int j = 0; j < 5; j++)
				str += (char) ch++;

			try {
				str = ex.exchange(str);
			} catch (InterruptedException exc) {
				System.out.println(exc);
			}
		}
	}
}

/**
 * 项目组B
 * @author liuyan
 */
class TeamB implements Runnable {
	Exchanger<String> ex;

	String str;

	TeamB(Exchanger<String> c) {
		ex = c;
		new Thread(this).start();
	}

	public void run() {

		for (int i = 0; i < 3; i++) {
			try {
				str = ex.exchange(new String());
				System.out.println("Got: " + str);
			} catch (InterruptedException exc) {
				System.out.println(exc);
			}
		}
	}
}

 

需要说明的就是这种交换一定是成对儿的，就是交换的线程数目一定是偶数。否则奇数个线程，剩下那一个和谁交换去？这也是“等价交易”的一种体现。

8.  总结

这次主要介绍了并发环境下常常使用的并发包，用于控制多线程的并发调度、同步、交互、交换、协作等等。使用这些协作同步器，可以更灵活的处理线程之间的关系。也能更好地使用硬件资源为我们的并发系统提供高效率的运行能力。当然这次总结仅仅限于使用的层次，底层的实现源码分析有时间再做总结。