Java线程类二

一、java.util.concurrent.Exchanger

Java 5中新增加了一个Exchanger类，这个类可以用来在一对线程之间交换元素，并且这种交换是线程安全的，不需要同步，具体说来就是每个线程将它想交换的对象放到exchanger对象中去，然后从这个对象返回对方线程用来交换的对象。有一点要求就是这两个交换的对象类型必须相同。

例如要实现生产者、消费者应用，以前可能的一种作法就是用一个集合：一个线程往里写，另一个线程从里面读。现在，如果用Exchanger的方式，那可以用两个集合，一个用在生产端，一个用在消费端，然后不时的对他们进行交换，例如：

		Exchanger<List<String>> exchanger = new Exchanger<List<String>>();
		Producer producer = new Producer(exchanger);
		Consumer consumer = new Consumer(exchanger);
		producer.start();
		consumer.start();

首先创建一个Exchanger对象，并指定了它将交换的内容格式，然后分别创建Producer对象和Consumer对象，并将此Exchanger对象传入。Producer类实现如下：

class Producer extends Thread{
	private Exchanger<List<String>> exchanger;
	
	private List<String> storage = new ArrayList<String>();
	
	public Producer(Exchanger<List<String>> e){
		this.exchanger = e;
	}
	
	@Override
	public void run() {
		int i = 0;
		while(true){
			//store something into storage
			storage.add("One"+i);
			storage.add("two"+i);
			storage.add("three"+i);
			try {
				//show the storage before exchange
				System.out.println("Produced "+storage);
				//exchange with consumer, and get the exchanged from it
				storage = exchanger.exchange(storage);
				//show the exchanged result
				System.out.println("After exchanged on Producer: "+storage);
				//sleep for a while before continue
				sleep(2000);
				i++;
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}
	
}

 它会不间断的生产，并且每次生产的内容都不一样，然后将生产的东西进行交换。它并不需要知道有谁在和它交换，只要将要交换的内容传给Exchanger，并从Exchanger取会交换的结果。Consumer的实现如下：

class Consumer extends Thread{
	private Exchanger<List<String>> exchanger;
	
	private List<String> storage = new ArrayList<String>();
	
	public Consumer(Exchanger<List<String>> e){
		this.exchanger = e;
	}
	
	@Override
	public void run() {
		while(true){
			try {
				/*
				 * do exchange, the storage is empty before exchanged
				 * and will be exchanged to producer
				 */
				storage = exchanger.exchange(storage);
				//show the exchanged result
				System.out.println("Consumed: "+storage);
				System.out.println("======================================");
				/*
				 * remember to reset storage before continue, else the values
				 * will be back to producer side
				 */
				storage.clear();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}
	
}

 执行过程和Producer类似，不过需要注意的是：在每次使用完交换来的内容后，要记得清空，要不然结果又会被交换回Producer端。以上程序某次运行的结果可能如下：

Produced [One0, two0, three0]
After exchanged on Producer: []
Consumed: [One0, two0, three0]
======================================
Produced [One1, two1, three1]
After exchanged on Producer: []
Consumed: [One1, two1, three1]
======================================
Produced [One2, two2, three2]
After exchanged on Producer: []
Consumed: [One2, two2, three2]
======================================

 二、分支/合并（Fork/Join)框架

Java 7提供了一个分支/合并(Fork/Join)框架，它的作者是Doug Lea，实现来源于他于2000年发表的一篇论文：http://gee.cs.oswego.edu/dl/papers/fj.pdf 。简单的说就是：Fork -- 将大任务分解成一系列小任务分别执行，每个小任务由一个线程执行; Join -- 在每个小任务执行完成后，将它们的结果合并，得到最终的结果。（我不太清楚它和最近比较流行的Mapper/Reduce编程模式有多大的不同）

 

要使用Fork/Join框架，涉及到以下几个主要的类：

1. ForkJoinPool：这是ForkJoin任务执行线程所在的线程池，由它创建一系统的线程去执行每个小任务。可以指定初始的并发数，如果不指定，由默认值为当前机器的内核数，这个可以从源码里得出：Runtime.getRuntime().availableProcessors()。
2. ForkJoinTask：这是ForkJoin任务的抽象实现，一般不直接继承它，而是使用它的以下两个抽象子类。
3. RecursiveAction：ForkJoinTask的子类，一般用于没有返回值的情况
4. RecursiveTask：ForkJoinTask的子类，一般用于有返回值的情况

 其中ForkJoinPool有三个用来提交或开始任务的方法：

1. submit：异步的执行Task，并且在任务结束后，可以使用getRawResult()方法获取返回值；在获取返回值之前，可能需要使用如 is*() 方法判断当前任务的执行结果。
2. execute：同submit类似，但是不带返回值
3. invoke：同submit类似，但是是同步的，即方法会在任务结果里才返回。可以看它的源码实现，返回task.join()或者task.invoke()。

而ForkJoinTask也有两个用来执行的方法：

1. fork：类似于上面的submit
2. invoke：类似于上面的invoke方法

下面用一个示例来介绍Fork/Join方法的用法。

我们都知道斐波那契数列由以下公式定义：

f(n) = f(n-1) + f(n-2); n-2>=0; f(0)=0; f(1)=1;

由以上对Fork/Join框架的介绍，可以试着用它来计算斐波那契数列的值：每一个f(n)都可以分解成对f(n-1)和f(n-2)的计算，然后把他们的结果合并，得到最终的值。

首先看一下Task的代码，如下：

	private static class FibonacciForkTask extends RecursiveTask<Long> { //因为需要有返回值，所以继承自RecursiveTask

		private static final long serialVersionUID = 1L;
		private int n;

		public FibonacciForkTask(int n) {
			this.n = n;
		}

		@Override
		protected Long compute() {
			if (n == 0) {
				return 0L;
			}
			if (n == 1) {
				return 1L;
			}
			FibonacciForkTask t1 = new FibonacciForkTask(n - 1);//子任务1
			FibonacciForkTask t2 = new FibonacciForkTask(n - 2);//子任务2
			invokeAll(t1, t2);//因为需要等待所有子任务的结果，所以用invokeAll()去调用
			try {
				Long long1 = t1.get();//取得结果1
				Long long2 = t2.get();//取得结果2
				return long1 + long2;//将两个子任务的结果求和，返回
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (ExecutionException e) {
				e.printStackTrace();
			}
			return null;
		}

	}

需要留意的两点：1.因为需要有返回值，所以子类继承RecursiveTask；因为结果需要把等待所有子任务完成然后求和，所以用invokeAll()调用。

然后我们看看怎么用这个子任务：

		ForkJoinPool forkJoinPool = new ForkJoinPool();
		ForkJoinTask<Long> submit = forkJoinPool.submit(new FibonacciForkTask(10));
		forkJoinPool.awaitTermination(0, TimeUnit.SECONDS);
		Long long1 = submit.get();
		System.out.println(long1);
		forkJoinPool.shutdown();

上面先创建一个ForkJoinPool对象，然后用submit()方法异常的调用之前我们定义的Task，然后等待任务结束，最后取得结果打印：

55

为了验证确实是有多线程在运行，并且进行了足够多次的任务划分，可以在compute()方法的任务划分之前加一个打印信息，例如（为了减少打印行数，我把上面的10改成了5）：

protected Long compute() {
			if (n == 0) {
				return 0L;
			}
			if (n == 1) {
				return 1L;
			}
			System.out.println(Thread.currentThread().getName());
                        ...
}

在我机器上打印的结果如下：

ForkJoinPool-1-worker-1
ForkJoinPool-1-worker-1
ForkJoinPool-1-worker-1
ForkJoinPool-1-worker-2
ForkJoinPool-1-worker-2
ForkJoinPool-1-worker-1
ForkJoinPool-1-worker-3
5

 可以看到执行了7次划分（可以手工计划一下，确实是需要进行7次划分），这些划分是在3个线程中执行的。如果把打印语句放到方法的首行，则有很多的打印结果，例如：

ForkJoinPool-1-worker-1
ForkJoinPool-1-worker-1
ForkJoinPool-1-worker-1
ForkJoinPool-1-worker-1
ForkJoinPool-1-worker-1
ForkJoinPool-1-worker-2
ForkJoinPool-1-worker-3
ForkJoinPool-1-worker-3
ForkJoinPool-1-worker-2
ForkJoinPool-1-worker-2
ForkJoinPool-1-worker-4
ForkJoinPool-1-worker-1
ForkJoinPool-1-worker-3
ForkJoinPool-1-worker-2
ForkJoinPool-1-worker-4
5

 可以看到总共有4个线程在运行（这也差不多说明了我的机器至少是4核的）。