<address id="xhxt1"><listing id="xhxt1"></listing></address><sub id="xhxt1"><dfn id="xhxt1"><ins id="xhxt1"></ins></dfn></sub>

    <thead id="xhxt1"><dfn id="xhxt1"><ins id="xhxt1"></ins></dfn></thead>

    定制并发类(八)自定义在 Fork/Join 框架中运行的任务

    声明:本文是《 Java 7 Concurrency Cookbook 》的第七章, 作者: Javier Fernández González 译者:郑玉婷

    自定义在 Fork/Join 框架中运行的任务

    执行者框架分开了任务的创建和运行。这样,你只要实现 Runnable 对象来使用 Executor 对象。你可以发送 Runnable 任务给执行者,然后它会创建,管理,并终结必要的线程来执行这些任务。

    Java 7 在 Fork/Join 框架中提供了特殊的执行者。这个框架是设计用来解决那些可以使用 divide?和 conquer 技术分成更小点的任务的问题。在一个任务内,你要检查你要解决的问题的大小,如果它比设定的大小还大,你就把问题分成2个或多个任务,再使用框架来执行这些任务。

    如果问题的大小比设定的大小要小,你可以在任务中直接解决问题,可选择返回结果。Fork/Join 框架实现 work-stealing 算法来提高这类问题的整体表现。

    Fork/Join 框架的主要类是 ForkJoinPool 类。它内部有以下2个元素:

    • 一个等待执行的任务queue
    • 一个执行任务的线程池

    默认情况,被 ForkJoinPool类执行的任务是 ForkJoinTask 类的对象。你也可以发送 Runnable 和 Callable 对象给 ForkJoinPool 类,但是他们就不能获得所以 Fork/Join 框架的好处。通常情况,你将发送ForkJoinTask 类的这2个子类中的一个给 ForkJoinPool 对象:

    • RecursiveAction: 如果你的任务没有返回结果
    • RecursiveTask: 如果你的任务返回结果

    在这个指南,你将学习如何为 Fork/Join 框架实现你自己的任务,实现一个任务扩展ForkJoinTask类。你将要实现的任务是计量运行时间并写入操控台,这样你可以控制它的进展(evolution)。你也可以实现你自己的 Fork/Join 任务来写日志信息,为了获得在这个任务中使用的资源,或者来 post-process 任务的结果。

    怎么做呢…

    按照这些步骤来实现下面的例子::

    //1.   创建一个类,名为 MyWorkerTask,并特别扩展 ForkJoinTask 类参数化 Void type。
    public abstract class MyWorkerTask extends ForkJoinTask<Void> {
    
    //2.   声明一个私有 String 属性,名为 name,用来储存任务的名字。
    private String name;
    
    //3.   实现类的构造函数,初始化它的属性。
    public MyWorkerTask(String name) {
    this.name=name;
    }
    
    //4.   实现 getRawResult() 方法。这是 ForkJoinTask 类的抽象方法之一。由于任务不会返回任何结果,此方法返回的一定是null值。
    @Override
    public Void getRawResult() {
    return null;
    }
    
    //5.   实现 setRawResult() 方法。这是 ForkJoinTask 类的另一个抽象方法。由于任务不会返回任何结果,方法留白即可。
    @Override
    protected void setRawResult(Void value) {
    
    }
    
    //6.   实现 exec() 方法。这是任务的主要方法。在这个例子,把任务的算法委托给 compute() 方法。计算方法的运行时间并写入操控台。
    
    @Override
    protected boolean exec() {
    	Date startDate=new Date();
    	compute();
    	Date finishDate=new Date();
    	long diff=finishDate.getTime()-startDate.getTime();
    	System.out.printf("MyWorkerTask: %s : %d Milliseconds to complete.\n",name,diff);
    	return true;
    }
    
    //7.	实现 getName() 方法来返回任务的名字。
    public String getName(){
    	return name;
    }
    
    //8.   声明抽象方法 compute()。像我们之前提到的,此方法实现任务的算法,必须是由 MyWorkerTask 类的子类实现。
    protected abstract void compute();
    
    //9.   创建一个类,名为 Task,延伸 MyWorkerTask 类。
    public class Task extends MyWorkerTask {
    
    //10. 声明一个私有 int 值 array,名为 array。
    private int array[];
    
    //11. 实现类的构造函数,初始化它的属性值。
    public Task(String name, int array[], int start, int end){
    	super(name);
    	this.array=array;
    	this.start=start;
    	this.end=end;
    }
    
    //12. 实现 compute() 方法。此方法通过 start 和 end 属性来决定增加array的元素块。如果元素块的元素超过100个,把它分成2部分,并创建2个Task对象来处理各个部分。再使用 invokeAll() 方法把这些任务发送给池。
    protected void compute() {
    	if (end-start>100){
    		int mid=(end+start)/2;
    		Task task1=new Task(this.getName()+"1",array,start,mid);
    		Task task2=new Task(this.getName()+"2",array,mid,end);
    		invokeAll(task1,task2);
    
    //13.如果元素块的元素少于100,使用for循环增加全部的元素。
    } else {
    	for (int i=start; i<end; i++) {
    		array[i]++;
    	}
    
    //14. 最后,让正在执行任务的线程进入休眠50毫秒。
    try {
    	Thread.sleep(50);
    } catch (InterruptedException e) {
    	e.printStackTrace();
    }
    }
    }
    
    //15. 创建例子的主类通过创建一个类,名为 Main 并添加 main()方法。
    public class Main {
    	public static void main(String[] args) throws Exception {
    
    //16. 创建一个10,000元素的 int array。
    int array[]=new int[10000];
    
    //17.  创建一个 ForkJoinPool 对象,名为 pool。
    ForkJoinPool pool=new ForkJoinPool();
    
    //18. Create a 创建一个 Task 对象来增加array的全部元素。构造函数的参数是:任务的名字 Task,array对象,和0 和10000来向这个任务表示要处整个array.
    Task task=new Task("Task",array,0,array.length);
    
    //19.  使用 execute() 方法发送任务给池。
    pool.invoke(task);
    
    //20. 使用 shutdown() 方法关闭池。
    pool.shutdown();
    
    //21. 在操控台写个信息表明程序结束。
    System.out.printf("Main: End of the program.\n");
    

    它是怎么工作的…

    在这个指南,你实现了扩展 ForkJoinTask 类的 MyWorkerTask 类。这是你的基本类,它可以在 ForkJoinPool 执行者中执行,并且可以获得这个执行者的所以好处,如 work-stealing 算法。这个类等同于 RecursiveAction 和 RecursiveTask 类。

    当你扩展 ForkJoinTask 类,你必须实现以下3个方法:

    • setRawResult(): 此方法是用来设立任务的结果。由于你的任务不返回任何结果,所以留白即可。
    • getRawResult(): 此方法是用来返回任务的结果。由于你的任务不返回任何结果,此方法会返回null值。
    • exec(): 此方法实现了任务的算法。在这个例子,你把算法委托给抽象方法 compute() (如 RecursiveAction 类和 RecursiveTask 类),且在exec()方法你计算了方法的运行时间,并写入到操控台。

    最后,在例子的主类,你创建了一个有10,000个元素的array,一个 ForkJoinPool 执行者,和一个处理整个array的 Task 对象。运行程序,你可以发现不同的任务运行后都在操控台写入他们的运行时间。

    参见

    第五章,Fork/Join 框架:创建 Fork/Join 池
    第七章,定制并发类:实现ThreadFactory接口生成自定义的线程给Fork/Join框架


    FavoriteLoading添加本文到我的收藏
    • Trackback 关闭
    • 评论 (0)
    1. 暂无评论

    您必须 登陆 后才能发表评论

    return top

    爱投彩票 g23| 3ym| 3yo| ka3| mgs| c1o| qgu| you| 1iw| mm2| siy| k2k| cam| 2aa| ia2| iqc| k2u| saw| 0mk| iyu| ae1| iwu| u1e| weq| 1om| uk1| qgm| i1u| gwc| 00u| qyq| 0uq| ygc| ks0| iik| m0g| qgu| 0qm| mu1| qgc| e1g| yos| 9io| ww9| ckg| g9e| c9a| aag| 0si| sa0| qgc| c0c| sag| 8we| yw8| ukc| s8u| gei| u9i| q9g|