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

Question

我正在阅读 Java ForkJoin 框架。不直接在 ForkJoinTask 的实现上调用 invoke()(例如 RecursiveTask)，而是实例化 ForkJoinPool 有什么额外的好处 并调用 pool.invoke(task)？当我们将这 2 个方法都称为 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

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

fork join pool 上的 invoke 方法向池提交任务，然后池开始在单独的线程上运行，调用该线程上的 compute 方法，然后然后等待结果。

您可以在 RecursiveTask 和 ForkJoinPool 的 java 文档中的措辞中看到这一点。 invoke() 方法实际执行任务，而 compute() 方法只是封装计算。

protected abstract V compute()

The main computation performed by this task.

和ForkJoinPool

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

Performs the given task, returning its result upon completion. ...

因此，对于计算方法，您正在做的是在 fork 连接池之外运行对 compute 的第一次调用。您可以通过在计算方法中添加日志行来对此进行测试。

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

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

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

一旦您调用了 invokeAll，该调用中包含的子任务就会在池中运行。但请注意，它不一定在调用 compute() 之前创建的池中运行。您可以注释掉您的 new ForkJoinPool() 代码，它仍然会运行。有趣的是，java 7 文档说如果在池管理线程之外调用 invokeAll() 方法将抛出异常，但 java 8 文档不会。请注意，我没有在 Java 7 中测试过它(只有 8 个)。但很有可能，当在 Java 7 中直接调用 compute() 时，您的代码会抛出异常。

两个结果同时返回的原因是毫秒不够准确，无法记录在池管理线程中启动第一个线程或仅运行第一个 compute 调用的区别在现有线程中。

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

pool.invoke() 调用在特定池上调用；当第一次调用 task.invoke 或 task.invokeAll 时，而不是让框架自行创建。这意味着您可以将池重新用于新任务，并在创建池时指定 Activity 线程数等内容。这就是区别。将这些日志行添加到您的代码中，尝试一下，您就会看到它在做什么