ForkJoinPool相同代码不同风格的延迟不同

Question

我试图将paralleStream与自定义的ForkJoin池一起使用，该任务执行网络调用。当我使用以下样式时

pool.submit(() -> {
        ioDelays.parallelStream().forEach(n -> {
            induceRandomSleep(n);
        });
    }).get();

for (final Integer num : ioDelays) {
        ForkJoinTask<Integer> task =  pool.submit(() -> {
            return induceRandomSleep(num);
        });
        tasks.add(task);
    }
    int count = 0;
    final List<Integer> returnVals = new ArrayList<>();
    tasks.forEach(task -> {
        try {
            returnVals.add(task.get());
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (ExecutionException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    });

如果使用parallelStream,那么ForkJoinPool.Common是否以某种方式参与其中？下面是模拟上述两种样式的整个程序

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;

public class FJTPExperiment {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ForkJoinPool pool = new ForkJoinPool(200);

        List<Integer> ioDelays = new ArrayList<>();
        for (int i = 0; i <2000; i++) {
            ioDelays.add( (int)(300 *Math.random() + 200));
        }
        int originalCount = 0;
        for (Integer val : ioDelays) {
            originalCount += val;
        }
        System.out.println("Expected " + originalCount);
        System.out.println(Thread.currentThread().getName() + " ::::Number of threads in common pool :" + ForkJoinPool.getCommonPoolParallelism());


        long beginTimestamp = System.currentTimeMillis();
        pool.submit(() -> {
            ioDelays.parallelStream().forEach(n -> {
                induceRandomSleep(n);
            });
        }).get();
        long endTimestamp = System.currentTimeMillis();
        System.out.println("Took " + (endTimestamp - beginTimestamp) + " ms");


        List<ForkJoinTask<Integer>> tasks = new ArrayList<>();
        beginTimestamp = System.currentTimeMillis();
        for (final Integer num : ioDelays) {
            ForkJoinTask<Integer> task =  pool.submit(() -> {
                return induceRandomSleep(num);
            });
            tasks.add(task);
        }
        int count = 0;
        final List<Integer> returnVals = new ArrayList<>();
        tasks.forEach(task -> {
            try {
                returnVals.add(task.get());
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            } catch (ExecutionException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        });
        endTimestamp = System.currentTimeMillis();
        for (Integer val : returnVals) {
            count += val;
        }
        System.out.println("Count " + count);
        System.out.println("Took " + (endTimestamp - beginTimestamp) + " ms");
    }


    public static int induceRandomSleep(int sleepInterval) {
        System.out.println(Thread.currentThread().getName() + " ::::sleeping for " + sleepInterval + " ms");
        try {
            Thread.sleep(sleepInterval);
            return sleepInterval;
        } catch (InterruptedException e) {
            e.printStackTrace();
            return sleepInterval;
        }
    }
}

Answer 1

我最终找到了答案问题有两个部分：

1）只有一个任务被提交给ForkJoinPool,它是如何产生多个线程的？

查看JDK实现，似乎在调用parallelStream时，它会检查当前线程是否是ForkJoinWorkerThread，如果是，则任务会被推送到客户ForkJoinPool的队列中，如果不是，则会被推送到ForkJoinPool.Common。这也通过日志进行了验证。

2）如果它能工作，为什么它慢？

因为并行性不是从自定义ForkJoinPool的并行性派生的，而是从ForkJoinPool.Common的并行性派生的，后者默认情况下限制为CPU核心数-1。这里是JDK实现，leaf_target是在这里派生的。如果这必须正常工作，那么应该有一个分支，它从自定义线程池的并行性派生leaf_target