ForkJoinPool.invoke()和ForkJoinTask.invoke()或compute()

Question

我正在阅读Java ForkJoin框架.通过不直接调用(例如)invoke()的实现,但实例化和调用有什么额外的好处？当我们称这两种方法全部被调用时到底发生了什么？ForkJoinTaskRecursiveTaskForkJoinPoolpool.invoke(task)invoke

从源代码来看,似乎如果recursiveTask.invoke被调用,它将以托管线程池的方式调用它exec并最终调用它compute.因此,为什么我们有这个成语更令人困惑pool.invoke(task).

我写了一些简单的代码来测试性能差异,但我没有看到任何.也许测试代码错了？见下文:

public class MyForkJoinTask extends RecursiveAction {

    private static int totalWorkInMillis = 20000;
    protected static int sThreshold = 1000;

    private int workInMillis;


    public MyForkJoinTask(int work) {
        this.workInMillis = work;
    }

    // Average pixels from source, write results into destination.
    protected void computeDirectly() {
        try {

            ForkJoinTask<Object> objectForkJoinTask = new ForkJoinTask<>();
            Thread.sleep(workInMillis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    protected void compute() {
        if (workInMillis < sThreshold) {
            computeDirectly();
            return;
        }

        int discountedWork = (int) (workInMillis * 0.9);
        int split = discountedWork / 2;

        invokeAll(new MyForkJoinTask(split),
                new MyForkJoinTask(split));
    }

    public static void main(String[] args) throws Exception {
        System.out.printf("Total work is %d in millis.%n", totalWorkInMillis);
        System.out.printf("Threshold is %d in millis.%n", sThreshold);

        int processors = Runtime.getRuntime().availableProcessors();
        System.out.println(Integer.toString(processors) + " processor"
                + (processors != 1 ? "s are " : " is ")
                + "available");

        MyForkJoinTask fb = new MyForkJoinTask(totalWorkInMillis);

        ForkJoinPool pool = new ForkJoinPool();

        long startTime = System.currentTimeMillis();


        // These 2 seems no difference!
        pool.invoke(fb);
//        fb.compute();


        long endTime = System.currentTimeMillis();

        System.out.println("Took " + (endTime - startTime) +
                " milliseconds.");
    }
}

Answer 1

类的compute()方法RecursiveTask只是一个包含任务代码的抽象方法.它不使用池中的新线程,如果您正常调用它,它不会在池托管线程中运行.

在invoke上叉方法加入池提交到池中,然后启动一个单独的线程中运行,调用任务compute在该线程的方法,然后等待结果.

你可以在java doc for RecursiveTask和ForkJoinPool的措辞中看到这一点.该invoke()方法实际执行任务,而该compute()方法只是封装计算.

protected abstract V compute()

该任务执行的主要计算.

和ForkJoinPool

public <T> T invoke(ForkJoinTask<T> task)

执行给定任务,完成后返回结果....

因此,使用compute方法,您正在执行的是compute对fork连接池外部的第一次调用.您可以通过在compute方法中添加日志行来测试它.

System.out.println(this.inForkJoinPool());

您还可以通过记录线程ID来检查它是否在同一个线程中运行

System.out.println(Thread.currentThread().getId());

拨打电话后invokeAll,该呼叫中包含的子任务将在池中运行.但请注意,它不一定在您调用之前创建的池中运行compute().您可以注释掉您的new ForkJoinPool()代码,它仍然可以运行.有趣的是,java 7 doc说invokeAll()如果在池托管线程之外调用它,该方法将抛出异常,但java 8 doc不会.我没有在java 7中测试过它(只有8个).但很可能,compute()在java 7中直接调用时,您的代码会抛出异常.

两个结果同时返回的原因是,毫秒不足以准确记录在池托管线程中启动第一个线程的差异,或者只是compute在现有线程中运行第一个调用.

Sierra和Bates建议您使用fork join框架的OCA/OCP学习指南的方法是invoke()从池中调用.它清楚地表明您正在使用哪个池,这也意味着您可以将多个任务提交到同一个池,这样可以节省每次重新创建新池的开销.从逻辑上讲,将所有任务计算保存在池托管线程中(或者至少我认为是这样)也更清晰.