9

我正在使用 CompletableFuture 异步执行从列表源生成的流。

所以我正在测试重载方法，即 CompletableFuture 的“supplyAsync”，其中一种方法仅采用单个供应商参数，而另一种采用供应商参数和执行器参数。这是两者的文档：

一

supplyAsync（供应商供应商）

返回一个新的 CompletableFuture，它由在 ForkJoinPool.commonPool() 中运行的任务异步完成，其值通过调用给定的供应商获得。

第二

supplyAsync(Supplier 供应商，Executor 执行者)

返回一个新的 CompletableFuture，它由在给定执行程序中运行的任务异步完成，其值通过调用给定供应商获得。

这是我的测试课：

public class TestCompleteableAndParallelStream {

    public static void main(String[] args) {
        List<MyTask> tasks = IntStream.range(0, 10)
                .mapToObj(i -> new MyTask(1))
                .collect(Collectors.toList());
        
        useCompletableFuture(tasks);
        
        useCompletableFutureWithExecutor(tasks);

    }
    
    public static void useCompletableFutureWithExecutor(List<MyTask> tasks) {
          long start = System.nanoTime();
          ExecutorService executor = Executors.newFixedThreadPool(Math.min(tasks.size(), 10));
          List<CompletableFuture<Integer>> futures =
              tasks.stream()
                   .map(t -> CompletableFuture.supplyAsync(() -> t.calculate(), executor))
                   .collect(Collectors.toList());
         
          List<Integer> result =
              futures.stream()
                     .map(CompletableFuture::join)
                     .collect(Collectors.toList());
          long duration = (System.nanoTime() - start) / 1_000_000;
          System.out.printf("Processed %d tasks in %d millis\n", tasks.size(), duration);
          System.out.println(result);
          executor.shutdown();
        }
    
    public static void useCompletableFuture(List<MyTask> tasks) {
          long start = System.nanoTime();
          List<CompletableFuture<Integer>> futures =
              tasks.stream()
                   .map(t -> CompletableFuture.supplyAsync(() -> t.calculate()))
                   .collect(Collectors.toList());
         
          List<Integer> result =
              futures.stream()
                     .map(CompletableFuture::join)
                     .collect(Collectors.toList());
          long duration = (System.nanoTime() - start) / 1_000_000;
          System.out.printf("Processed %d tasks in %d millis\n", tasks.size(), duration);
          System.out.println(result);
        }
    
    

}


class MyTask {
      private final int duration;
      public MyTask(int duration) {
        this.duration = duration;
      }
      public int calculate() {
        System.out.println(Thread.currentThread().getName());
        try {
          Thread.sleep(duration * 1000);
        } catch (final InterruptedException e) {
          throw new RuntimeException(e);
        }
        return duration;
      }
    }

“useCompletableFuture”方法大约需要 4 秒才能完成，而“useCompletableFutureWithExecutor”方法只需 1 秒即可完成。

不，我的问题是，可以做开销的 ForkJoinPool.commonPool() 有什么不同的处理？那我们不应该总是更喜欢自定义执行器池而不是 ForkJoinPool 吗？

4

2 回答 2

8

检查ForkJoinPool.commonPool()尺寸。默认情况下，它创建一个大小为

Runtime.getRuntime().availableProcessors() - 1

我在我的 Intel i7-4800MQ（4 个核心 + 4 个虚拟核心）上运行您的示例，在我的情况下，公共池的大小为7，因此整个计算耗时约 2000 毫秒：

ForkJoinPool.commonPool-worker-1
ForkJoinPool.commonPool-worker-4
ForkJoinPool.commonPool-worker-2
ForkJoinPool.commonPool-worker-6
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-4
ForkJoinPool.commonPool-worker-2
ForkJoinPool.commonPool-worker-1
Processed 10 tasks in 2005 millis
[1, 1, 1, 1, 1, 1, 1, 1, 1, 1]

在第二种情况下，您使用

Executors.newFixedThreadPool(Math.min(tasks.size(), 10));

所以池中有 10 个线程准备好执行计算，所以所有任务都在 ~1000 毫秒内运行：

pool-1-thread-1
pool-1-thread-2
pool-1-thread-3
pool-1-thread-4
pool-1-thread-5
pool-1-thread-6
pool-1-thread-7
pool-1-thread-8
pool-1-thread-9
pool-1-thread-10
Processed 10 tasks in 1002 millis
[1, 1, 1, 1, 1, 1, 1, 1, 1, 1]

ForkJoinPool和之间的区别ExecutorService

尤金在他的评论中还提到了一件更重要的事情。ForkJoinPool使用工作窃取方法：

ForkJoinPool与其他类型的 ExecutorService的不同之处主要在于采用了工作窃取：池中的所有线程都尝试查找并执行提交到池和/或由其他活动任务创建的任务（如果不存在，则最终阻塞等待工作） . 当大多数任务产生其他子任务（大多数 ForkJoinTasks 也是如此）时，以及当许多小任务从外部客户端提交到池时，这可以实现高效处理。尤其是在构造函数中将 asyncMode 设置为 true 时，ForkJoinPools 也可能适用于从未加入的事件式任务。

而ExecutorService创建时.newFixedThreadPool()使用分而治之的方法。

如何确定池大小？

有一个关于什么是最佳线程池大小的问题，您可以在那里找到有用的信息：

设置理想的线程池大小

这个线程也是一个调查的好地方：

Java 8 并行流中的自定义线程池

于 2017-08-02T12:25:45.613 回答
5

进一步检查互联网上的解决方案，我发现我们可以使用以下属性更改 ForkJoinPool 采用的默认池大小：

-Djava.util.concurrent.ForkJoinPool.common.parallelism=16

因此，此属性可以进一步帮助以更有效的方式和更多的并行性使用 ForkJoinPool。

于 2017-08-02T12:33:46.393 回答