`
wankunde
  • 浏览: 162848 次
  • 性别: Icon_minigender_1
  • 来自: 南京
社区版块
存档分类
最新评论

非主流并发工具之 ForkJoinPool

 
阅读更多

 

ForkJoinPool 是 Java SE 7 新功能“分叉/结合框架”的核心类,现在可能乏人问津,但我觉得它迟早会成为主流。分叉/结合框架是一个比较特殊的线程池框架,专用于需要将一个任务不断分解成子任务(分叉),再不断进行汇总得到最终结果(结合)的计算过程。比起传统的线程池类ThreadPoolExecutorForkJoinPool 实现了工作窃取算法,使得空闲线程能够主动分担从别的线程分解出来的子任务,从而让所有的线程都尽可能处于饱满的工作状态,提高执行效率。

ForkJoinPool 提供了三类方法来调度子任务:

execute 系列
异步执行指定的任务。
invoke 和 invokeAll
执行指定的任务,等待完成,返回结果。
submit 系列
异步执行指定的任务并立即返回一个 Future 对象。

子任务由 ForkJoinTask 的实例来代表。它是一个抽象类,JDK 为我们提供了两个实现:RecursiveTask 和 RecursiveAction,分别用于需要和不需要返回计算结果的子任务。ForkJoinTask 提供了三个静态的 invokeAll 方法来调度子任务,注意只能在 ForkJoinPool 执行计算的过程中调用它们。

ForkJoinPool 和 ForkJoinTask 还提供了很多让人眼花缭乱的公共方法,其实它们大多数都是其内部实现去调用的,对于应用开发人员来说意义不大。

下面以统计 D 盘文件个数为例。这实际上是对一个文件树的遍历,我们需要递归地统计每个目录下的文件数量,最后汇总,非常适合用分叉/结合框架来处理:

 

 

// 处理单个目录的任务
public class CountingTask extends RecursiveTask<Integer> {
    private Path dir;
 
    public CountingTask(Path dir) {
        this.dir = dir;
    }
 
    @Override
    protected Integer compute() {
        int count = 0;
        List<CountingTask> subTasks = new ArrayList<>();
 
        // 读取目录 dir 的子路径。
        try (DirectoryStream<Path> ds = Files.newDirectoryStream(dir)) {
            for (Path subPath : ds) {
                if (Files.isDirectory(subPath, LinkOption.NOFOLLOW_LINKS)) {
                    // 对每个子目录都新建一个子任务。
                    subTasks.add(new CountingTask(subPath));
                } else {
                    // 遇到文件,则计数器增加 1。
                    count++;
                }
            }
 
            if (!subTasks.isEmpty()) {
                // 在当前的 ForkJoinPool 上调度所有的子任务。
                for (CountingTask subTask : invokeAll(subTasks)) {
                    count += subTask.join();
                }
            }
        } catch (IOException ex) {
            return 0;
        }
        return count;
    }
}
 
// 用一个 ForkJoinPool 实例调度“总任务”,然后敬请期待结果……
Integer count = new ForkJoinPool().invoke(new CountingTask(Paths.get("D:/")));

 

 

 

在我的笔记本上,经多次运行这段代码,耗费的时间稳定在 600 豪秒左右。普通线程池(Executors.newCachedThreadPool())耗时 1100 毫秒左右,足见工作窃取的优势。

结束本文前,我们来围观一个最神奇的结果:单线程算法(使用 Files.walkFileTree(...))比这两个都快,平均耗时 550 毫秒!这警告我们并非引入多线程就能优化性能,并须要先经过多次测试才能下结论。

 

 

转自:http://www.blogjava.net/shinzey/  

 

 

 

 

实际测试

 

package com.forkjoin.countdir;

import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

import com.util.DateTime;

public class CountTaskRecursive extends RecursiveTask {

	int sum =0;
	File file ;
	
	public CountTaskRecursive(File file) {
		this.file = file;
	}
	
	@Override
	protected Integer compute() {
		Integer csum =0;
		List<CountTaskRecursive> tasklist = new ArrayList<CountTaskRecursive>() ;
		if(file.isDirectory())
		{
			for(File f:file.listFiles())
			{
				CountTaskRecursive t = new CountTaskRecursive(f);
				tasklist.add(t);
			}
		}
		else
			csum ++;
		if(!tasklist.isEmpty())
		{
			for(CountTaskRecursive t :invokeAll(tasklist))
				csum += (Integer)t.join();
		}
		return csum;
	}

	public static void main(String[] args) {
		DateTime dt = new DateTime() ;
		System.out.println("系统日期:"+dt.getDate()) ;
		System.out.println("中文日期:"+dt.getDateComplete()) ;
		System.out.println("时间戳:"+dt.getTimeStamp()) ;
		CountTaskRecursive task = new CountTaskRecursive(new File("F:\\私人资料"));
		Integer sum = (Integer)new ForkJoinPool().invoke(task);
		System.out.println(sum);
		dt = new DateTime() ;
		System.out.println("系统日期:"+dt.getDate()) ;
		System.out.println("中文日期:"+dt.getDateComplete()) ;
		System.out.println("时间戳:"+dt.getTimeStamp()) ;
	}
}

 

package com.forkjoin.countdir;
import java.io.File;
import java.security.Timestamp;
import java.util.Calendar;

import com.util.DateTime;



public class CountTaskSingle {
	static int sum = 0;
	
	public void countDir(File file)
	{
		if(file.isDirectory())
		{
			for(File f:file.listFiles())
				countDir(f);
		}
		else
			sum++;
		
	}
	
	public static void main(String[] args) {
		CountTaskSingle ins = new CountTaskSingle();
		DateTime dt = new DateTime() ;
		System.out.println("系统日期:"+dt.getDate()) ;
		System.out.println("中文日期:"+dt.getDateComplete()) ;
		System.out.println("时间戳:"+dt.getTimeStamp()) ;
		ins.countDir(new File("F:\\私人资料"));
		System.out.println(sum);
		dt = new DateTime() ;
		System.out.println("系统日期:"+dt.getDate()) ;
		System.out.println("中文日期:"+dt.getDateComplete()) ;
		System.out.println("时间戳:"+dt.getTimeStamp()) ;
	}
}

 

 

package com.util;

import java.util.*; // 导入需要的工具包
public class DateTime { // 以后直接通过此类就可以取得日期时间
	private Calendar calendar = null; // 声明一个Calendar对象,取得时间
	public DateTime() { // 构造方法中直接实例化对象
		this.calendar = new GregorianCalendar();
	}
	public String getDate() { // 得到的是一个日期:格式为:yyyy-MM-dd HH:mm:ss.SSS
		// 考虑到程序要频繁修改字符串,所以使用StringBuffer提升性能
		StringBuffer buf = new StringBuffer();
		buf.append(calendar.get(Calendar.YEAR)).append("-"); // 增加年
		buf.append(this.addZero(calendar.get(Calendar.MONTH) + 1, 2)).append("-"); // 增加月
		buf.append(this.addZero(calendar.get(Calendar.DAY_OF_MONTH), 2)).append(" "); // 取得日
		buf.append(this.addZero(calendar.get(Calendar.HOUR_OF_DAY), 2)).append(":"); // 取得时
		buf.append(this.addZero(calendar.get(Calendar.MINUTE), 2)).append(":");
		buf.append(this.addZero(calendar.get(Calendar.SECOND), 2)).append(".");
		buf.append(this.addZero(calendar.get(Calendar.MILLISECOND), 3));
		return buf.toString();
	}
	public String getDateComplete() { // 得到的是一个日期:格式为:yyyy年MM月dd日 HH时mm分ss秒SSS毫秒
		// 考虑到程序要频繁修改字符串,所以使用StringBuffer提升性能
		StringBuffer buf = new StringBuffer();
		buf.append(calendar.get(Calendar.YEAR)).append("年"); // 增加年
		buf.append(this.addZero(calendar.get(Calendar.MONTH) + 1, 2)).append("月"); // 增加月
		buf.append(this.addZero(calendar.get(Calendar.DAY_OF_MONTH), 2)).append("日"); // 取得日
		buf.append(this.addZero(calendar.get(Calendar.HOUR_OF_DAY), 2)).append("时"); // 取得时
		buf.append(this.addZero(calendar.get(Calendar.MINUTE), 2)).append("分"); // 取得分
		buf.append(this.addZero(calendar.get(Calendar.SECOND), 2)).append("秒"); // 取得秒
		buf.append(this.addZero(calendar.get(Calendar.MILLISECOND), 3)).append("毫秒"); // 取得毫秒
		return buf.toString();
	}
	public String getTimeStamp() { // 得到的是一个时间戳
		// 考虑到程序要频繁修改字符串,所以使用StringBuffer提升性能
		StringBuffer buf = new StringBuffer();
		buf.append(calendar.get(Calendar.YEAR)); // 增加年
		buf.append(this.addZero(calendar.get(Calendar.MONTH) + 1, 2)); // 增加月
		buf.append(this.addZero(calendar.get(Calendar.DAY_OF_MONTH), 2)); // 取得日
		buf.append(this.addZero(calendar.get(Calendar.HOUR_OF_DAY), 2)); // 取得时
		buf.append(this.addZero(calendar.get(Calendar.MINUTE), 2)); // 取得分
		buf.append(this.addZero(calendar.get(Calendar.SECOND), 2)); // 取得秒
		buf.append(this.addZero(calendar.get(Calendar.MILLISECOND), 3)); // 取得毫秒
		return buf.toString();
	}
	// 考虑到日期中存在前导0,所以在此处加上补零的方法
	private String addZero(int num, int len) {
		StringBuffer s = new StringBuffer();
		s.append(num);
		while (s.length() < len) { // 如果长度不足,则继续补0
			s.insert(0, "0"); // 在第一个位置处补0
		}
		return s.toString();
	}
};

 

 

执行结果:

单线程执行结果

 

系统日期:2012-05-09 23:59:07.468

中文日期:2012年05月09日23时59分07秒468毫秒

时间戳:20120509235907468

389578

系统日期:2012-05-09 23:59:56.421

中文日期:2012年05月09日23时59分56秒421毫秒

时间戳:20120509235956421

 

多线程执行结果

 

系统日期:2012-05-10 00:01:57.500

中文日期:2012年05月10日00时01分57秒500毫秒

时间戳:20120510000157500

389578

系统日期:2012-05-10 00:09:00.281

中文日期:2012年05月10日00时09分00秒281毫秒

时间戳:20120510000900281

 

 

 

分享到:
评论
1 楼 宋建勇 2016-09-09  
示例不错,讲解也实在!受益了。

相关推荐

    13、线程池ForkJoinPool工作原理分析

    线程池ForkJoinPool是Java并发编程中的一个重要工具,它是Java 7引入的一个新特性,主要用于优化并行计算,特别是在处理大量可分任务时,如递归算法。ForkJoinPool是基于工作窃取(Work-Stealing)算法的线程池,...

    Java新的线程持池ForkJoinPool.pdf

    总之,ForkJoinPool是Java提供的一种高级并发工具,特别适合处理大量可以拆分的计算任务,通过工作窃取算法提高了并发效率。在编写高性能的并行计算程序时,理解并熟练运用ForkJoinPool是非常重要的。然而,使用时也...

    Java线程池ForkJoinPool实例解析

    Java线程池ForkJoinPool实例解析是Java并发编程中的一种高级主题,ForkJoinPool是Java 7中引入的一种新的线程池实现,它可以充分利用多CPU和多核CPU的优势,使得并发编程变得更加高效。 ForkJoinPool的优势在于,...

    Java多线程ForkJoinPool实例详解

    Java多线程编程中的ForkJoinPool实例详解是Java 7中引入的一种高效的并发编程框架。ForkJoinPool是ExecutorService接口的实现,它管理工作窃取算法(Work-Stealing Algorithm)实现高效的任务执行和线程管理。 ...

    13、线程池ForkJoinPool实战及其工作原理分析(1).pdf

    - **任务的分割与合并**: 在`ForkJoinPool`中,任务通过`ForkJoinTask`接口表示,并可以通过`ForkJoinPool`的`invoke()`方法提交。当一个任务被提交时,它会被分解成更小的任务,直到达到一定的阈值(通常称为基础...

    java并发工具包详解

    19. 使用 ForkJoinPool 进行分叉和合并 20. 锁 Lock 21. 读写锁 ReadWriteLock 22. 原子性布尔 AtomicBoolean 23. 原子性整型 AtomicInteger 24. 原子性长整型 AtomicLong 25. 原子性引用型 AtomicReference

    java7 并发编程forjoinpool

    Java 7 引入了一种新的并发编程框架——ForkJoinPool,它是基于分而治之(Divide and Conquer)策略的并行计算模型。ForkJoinPool 和与其配合使用的 RecursiveAction 和 RecursiveTask 类,为开发者提供了更高效地...

    Fork-Join框架演示

    - **RecusiveAction类**:介绍了`RecusiveAction`类的使用方法,包括创建子任务、调用`fork()`方法将任务异步提交给`ForkJoinPool`,以及`join()`方法等待所有子任务完成。 - **RecursiveTask类**:与`...

    java并发工具包 java.util.concurrent中文版用户指南pdf

    19. 使用 ForkJoinPool 进行分叉和合并 20. 锁 Lock 21. 读写锁 ReadWriteLock 22. 原子性布尔 AtomicBoolean 23. 原子性整型 AtomicInteger 24. 原子性长整型 AtomicLong 25. 原子性引用型 AtomicReference

    高性能并发业务.pdf

    Java并发的内置实现涉及到了Thread、Runnable、Callable等基础构建,以及ExecuterService/ThreadPool和ForkJoinPool等高级并发工具。ForkJoinPool是一种特殊的线程池,特别适合于有大量小任务需要并行处理的场景,...

    Java并发工具包java.util.concurrent用户指南中英文对照阅读版.pdf

    使用 ForkJoinPool 进行分叉和合并 20. 锁 Lock 21. 读写锁 ReadWriteLock 22. 原子性布尔 AtomicBoolean 23. 原子性整型 AtomicInteger 24. 原子性长整型 AtomicLong 25. 原子性引用型 AtomicReferenc

    java并发之并发工具类

    Java 并发工具类详解 Java 并发工具类是 Java 并发编程中的一种重要工具,用于协调线程之间的并发执行。 Java 并发工具类中有很多种,今天我们主要介绍四种:CyclicBarrier、CountDownLatch、Semaphore 和 ...

    Java并发工具包

    Java并发工具包是Java平台中的一个关键特性,它位于`java.util.concurrent`包下,为开发者提供了高效、安全的多线程编程支持。这个工具包的设计目标是简化并发编程,提高程序的性能和可维护性,同时避免了线程同步的...

    java并发工具类

    除了以上介绍的工具,Java并发工具类还包括`AtomicReference`, `AtomicIntegerArray`, `AtomicLongArray`, `AtomicMarkableReference`, `AtomicStampedReference`, `ThreadLocal`, `ForkJoinPool`等,它们各自在不同...

    内网并发测试工具

    内网并发测试工具是针对网络服务接口进行性能和稳定性测试的重要软件,特别是在内网环境中的应用,它能够帮助开发者和测试工程师模拟大量并发用户请求,以此来检验系统在高负载情况下的表现。这类工具通常包括多种...

    Java并发工具包java.util.concurrent用户指南中英文对照阅读版

    19. 使用 ForkJoinPool 进行分叉和合并 20. 锁 Lock 21. 读写锁 ReadWriteLock 22. 原子性布尔 AtomicBoolean 23. 原子性整型 AtomicInteger 24. 原子性长整型 AtomicLong 25. 原子性引用型 AtomicReference

    java并发工具包

    Java并发工具包是Java平台中用于处理多线程并行计算的重要组件,它包含在`java.util.concurrent`包中。这个工具包提供了多种高级并发工具,使得开发者能够编写出高效、安全、易于理解和维护的多线程代码。下面将详细...

    parallel-stream-fork-join-pool

    所有并行流执行都使用相同的(单例)线程池:ForkJoinPool.commonPool()。 这就是为什么在并行流中执行 IO(更常见的是阻塞调用)非常糟糕的原因:被阻塞的线程无法被 JVM 中的所有并行流使用。 为此,您必须改用 ...

    WEBAPI多线程并发测试工具

    标题"WEBAPI多线程并发测试工具"指出,这是一个专门针对Web API进行多线程并发测试的工具。Web API通常指的是应用程序接口,它们允许不同的服务之间进行通信,以实现数据交换和功能整合。多线程并发测试则是验证在多...

    Httpbfljcsgj_并发连接测试工具(内网版).rar

    "Httpbfljcsgj_并发连接测试工具(内网版).rar" 提供的正是这样一种专门用于测试网站并发连接性能的工具,适用于内网环境。 并发连接测试工具的主要作用是模拟多个用户同时访问网站或服务,以检查服务器在高负载情况...

Global site tag (gtag.js) - Google Analytics