欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

Java开发笔记(一百零六)Fork+Join框架实现分而治之

程序员文章站 2022-04-14 18:48:25
前面依次介绍了普通线程池和定时器线程池的用法,这两种线程池有个共同点,就是线程池的内部线程之间并无什么关联,然而某些情况下的各线程间存在着前因后果关系。譬如人口普查工作,大家都知道我国总人口为14亿左右,可是14亿的数目是怎么数出来呢?倘若只有一个人去统计,从小数到老都数不完。好比一个线程老牛破车干 ......

前面依次介绍了普通线程池和定时器线程池的用法,这两种线程池有个共同点,就是线程池的内部线程之间并无什么关联,然而某些情况下的各线程间存在着前因后果关系。譬如人口普查工作,大家都知道我国总人口为14亿左右,可是14亿的数目是怎么数出来呢?倘若只有一个人去统计,从小数到老都数不完。好比一个线程老牛破车干不了多少事情,既然如此,不妨多起一些线程呗。于是人口普查工作就由*分解到各个省份,各省又分派到下面的市县,再由市县分派到更下面的街道或乡镇,每个街道和乡镇统计完本辖区内的人口数量后,分别上报给对应的市县,市县再上报给省里,最后由各省上报*,这才统计完成全国的人口总数。在人口普查的案例中,这些线程不但存在上下级关系,而且下级线程的任务由上级线程分派而来,同时下级线程的处理结果又要交给上级线程汇总。根据任务流的走向,可将整个处理过程划分成下列三个阶段:
1、第一阶段从主线程开始,从上往下逐级分解任务,此时线程总数逐渐变多,每个分线程都先后收到上级线程分派的任务;
2、第二阶段由最下面的基层线程进行具体的任务操作,此时线程总数是不变的;
3、第三阶段从基层线程开始,从下往上逐级汇总任务结果,此时线程总数逐渐变少,最后主线程会收到汇总完成的最终结果;
以上的第一阶段,概括地说叫做“分而治之”;至于第三阶段,可概括称之为“汇聚归一”。为了实现这种分而治之的业务需求,java7新增了fork/join框架用以对症下药。该框架的fork操作会按照树状结构不断分出下级线程,其对应的是分而治之的过程;而join操作则把叶子线程的运算结果逐级合并,其对应的是汇聚归一的过程。在这分分合合的过程当中,悄然浮现出fork/join框架专用的线程池工具forkjoinpool,而它正是从executorservice派生出来的一个子类。鉴于分治策略的特殊性质,fork/join框架并不使用常见的runnable任务,而改为使用专门的递归任务recursivetask,该任务的fork方法实现了分而治之的fork操作,join方法实现了汇聚归一的join操作。
举个简单应用的例子,对于一段连续的数列求和,比如对0到99之间的所有整数求和,通常的做法是写个循环语句依次累加。常规的写法显然只有一个主线程在执行加法运算,无法体现多核cpu的性能优势,故而可以尝试将求和操作分而治之,先把整段数列划分为若干个子数列,再对各个子数列分别求和,最后汇总所有子数列的求和结果。采取recursivetask实现这种分派求和任务的话,可参见下面的代码例子,注意递归任务的入口由run方法改成了compute方法:

//定义一个求和的递归任务
public class sumtask extends recursivetask<integer> {
	private static final long serialversionuid = 1l;
	private static final int threshold = 20; // 不可再切割的元素个数门槛
	private int src[]; // 待求和的整型数组
	private int start; // 待求和的下标起始值
	private int end; // 待求和的下标终止值

	public sumtask(int[] src, int start, int end) {
		this.src = src;
		this.start = start;
		this.end = end;
	}

	// 对指定区间的数组元素求和
	private integer subtotal() {
		integer sum = 0;
		for (int i = start; i < end; i++) { // 求数组在指定区间的元素之和
			sum += src[i];
		}
		// 打印求和日志,包括当前线程的名称、起始数值、终止数值、区间之和
		string desc = string.format("%s ∑(%d~%d)=%d", thread.currentthread().getname(), start, end, sum);
		system.out.println(desc);
		return sum;
	}

	@override
	protected integer compute() {
		if ((end - start) <= threshold) { // 不可再切割了
			return subtotal(); // 对指定区间的数组元素求和
		} else { // 区间过大,还能继续切割
			int middle = (start + end) / 2; // 计算区间中线的位置
			// 创建左边分区的求和任务
			sumtask left = new sumtask(src, start, middle);
			left.fork(); // 把左边求和任务添加到处理队列中
			// 创建右边分区的求和任务
			sumtask right = new sumtask(src, middle, end);
			right.fork(); // 把右边求和任务添加到处理队列中
			// 左边子任务的求和结果加上右边子任务的求和结果,等于当前任务的求和结果
			int sum = left.join() + right.join();
			// 打印求和日志,包括当前线程的名称、起始数值、终止数值、区间之和
			string desc = string.format("%s ∑(%d~%d)=%d", thread.currentthread().getname(), start, end, sum);
			system.out.println(desc);
			return sum; // 返回本次任务的求和结果
		}
	}
}

然后外部往上面的求和任务输入待求和的整型数组,并调用任务对象的invoke获取执行结果,即可命令内置的线程池启动求和任务。调用代码示例如下:

	// 测试任务自带的线程池框架
	private static void testinternaltask() {
		// 下面初始化从0到99的整型数组
		int[] arr = new int[100];
		for (int i = 0; i < 100; i++) {
			arr[i] = i + 1;
		}
		// 创建一个求和的递归任务
		sumtask task = new sumtask(arr, 0, arr.length);
		try {
			// 执行同步任务,并返回执行结果。任务的invoke方法使用了内部的forkjoinpool
			integer result = task.invoke();
			system.out.println("最终计算结果: " + result);
		} catch (exception e) {
			e.printstacktrace();
		}
	}

 

运行以上的调用代码,输出下列的线程池日志:

forkjoinpool.commonpool-worker-3: ∑(0~12)=78
forkjoinpool.commonpool-worker-0: ∑(75~87)=978
forkjoinpool.commonpool-worker-2: ∑(50~62)=678
forkjoinpool.commonpool-worker-0: ∑(87~100)=1222
forkjoinpool.commonpool-worker-3: ∑(12~25)=247
forkjoinpool.commonpool-worker-3: ∑(0~25)=325
forkjoinpool.commonpool-worker-0: ∑(75~100)=2200
forkjoinpool.commonpool-worker-2: ∑(62~75)=897
forkjoinpool.commonpool-worker-2: ∑(50~75)=1575
forkjoinpool.commonpool-worker-1: ∑(37~50)=572
forkjoinpool.commonpool-worker-3: ∑(25~37)=378
forkjoinpool.commonpool-worker-3: ∑(25~50)=950
forkjoinpool.commonpool-worker-1: ∑(0~50)=1275
forkjoinpool.commonpool-worker-2: ∑(50~100)=3775
main: ∑(0~100)=5050
最终计算结果: 5050

 

从日志可见,fork/join框架的默认线程池一共启动了四个线程(正好是设备的cpu个数),同时最后一步的统计工作由主线程来完成。

注意到前述的调用代码并未写明fork/join框架的线程池工具forkjoinpool,这是因为递归任务拥有默认的内置线程池,即使外部不指定线程池对象,递归任务也会使用内置线程池进行线程调度。不过默认的线程池无法设置个性化的参数,所以还是建议在代码中显式指定forkjoinpool线程池,并调用线程池对象的execute/invoke/submit三个方法之一启动递归任务。有关这三个方法的具体用途说明如下:
execute:异步执行指定任务,且无返回值。
invoke:同步执行指定任务,并等待返回值,返回值就是最终的运算结果。
submit:异步执行指定任务,且返回结果任务对象。之后可择机调用结果任务的get方法获取最终的运算结果。
下面是在外部调用时显式指定线程池的求和代码例子:

	// 测试任务以外的线程池框架
	private static void testpooltask() {
		// 下面初始化从0到99的整型数组
		int[] arr = new int[100];
		for (int i = 0; i < 100; i++) {
			arr[i] = i + 1;
		}
		// 创建一个求和的递归任务
		sumtask task = new sumtask(arr, 0, arr.length);
		// 创建一个用于分而治之的线程池,并发数量为6
		forkjoinpool pool = new forkjoinpool(6);
		// 命令线程池执行求和任务,并返回存放执行结果的任务对象
		forkjointask<integer> taskresult = pool.submit(task);
		try {
			integer result = taskresult.get(); // 等待执行完成,并获取求和的结果数值
			system.out.println("最终计算结果: " + result);
		} catch (exception e) {
			e.printstacktrace();
		}
		pool.shutdown(); // 关闭线程池
	}

 

运行修改后的调用代码,输出下列的线程池日志:

forkjoinpool-1-worker-1: ∑(0~12)=78
forkjoinpool-1-worker-3: ∑(62~75)=897
forkjoinpool-1-worker-5: ∑(12~25)=247
forkjoinpool-1-worker-5: ∑(87~100)=1222
forkjoinpool-1-worker-5: ∑(25~37)=378
forkjoinpool-1-worker-5: ∑(37~50)=572
forkjoinpool-1-worker-5: ∑(25~50)=950
forkjoinpool-1-worker-1: ∑(0~25)=325
forkjoinpool-1-worker-4: ∑(50~62)=678
forkjoinpool-1-worker-4: ∑(50~75)=1575
forkjoinpool-1-worker-6: ∑(75~87)=978
forkjoinpool-1-worker-6: ∑(75~100)=2200
forkjoinpool-1-worker-2: ∑(0~50)=1275
forkjoinpool-1-worker-3: ∑(50~100)=3775
forkjoinpool-1-worker-1: ∑(0~100)=5050
最终计算结果: 5050

 

由日志可见,此时的线程池运行情况与刚才相比有两点不同:其一开启的线程数量变多了,这缘于新的线程池对象设置了并发数量为6;其二最后一步的统计工作仍在线程池内部执行,因而减轻了主线程的负担。结论当然是外部显式指定forkjoinpool的方式更优。



更多java技术文章参见《java开发笔记(序)章节目录