java8并行流

這篇文章展示了Java 8的CompletableFuture在執行異步計算時如何與并行流進行比較。

我們將使用以下類對長時間運行的任務進行建模：

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;} }

讓我們創建十個任務，每個任務持續1秒：

List<MyTask> tasks = IntStream.range(0, 10).mapToObj(i -> new MyTask(1)).collect(toList());

我們如何有效地計算任務清單？

方法1：依次

您首先想到的是按順序計算任務，如下所示：

public static void runSequentially(List<MyTask> tasks) {long start = System.nanoTime();List<Integer> result = tasks.stream().map(MyTask::calculate).collect(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); }

如您所料，這需要10秒鐘才能運行，因為每個任務都在main線程上一個接一個地運行。

方法2：使用并行流

快速改進是將您的代碼轉換為使用并行流，如下所示：

public static void useParallelStream(List<MyTask> tasks) {long start = System.nanoTime();List<Integer> result = tasks.parallelStream().map(MyTask::calculate).collect(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); }

輸出是

main ForkJoinPool.commonPool-worker-1 ForkJoinPool.commonPool-worker-3 ForkJoinPool.commonPool-worker-2 ForkJoinPool.commonPool-worker-3 ForkJoinPool.commonPool-worker-2 main ForkJoinPool.commonPool-worker-1 ForkJoinPool.commonPool-worker-1 main Processed 10 tasks in 3043 millis

這次花了3秒，因為并行執行了4個任務（使用了來自ForkJoinPool三個線程以及main線程）。

方法3：使用CompletableFutures

讓我們看看CompletableFuture的性能是否更好：

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); }

在上面的代碼中，我們首先獲取CompletableFuture的列表，然后在每個CompletableFuture調用join方法以等待它們CompletableFuture完成。 請注意， join與get相同，唯一的區別是前者不引發任何檢查的異常，因此在lambda表達式中更為方便。

另外，您必須使用兩個單獨的流管道，而不是將兩個map操作彼此放在后面，因為中間流操作是惰性的，您將不得不按順序處理任務！ 這就是為什么您首先需要在列表中收集CompletableFuture ，以允許它們在等待完成之前啟動。

輸出是

ForkJoinPool.commonPool-worker-1 ForkJoinPool.commonPool-worker-2 ForkJoinPool.commonPool-worker-3 ForkJoinPool.commonPool-worker-1 ForkJoinPool.commonPool-worker-2 ForkJoinPool.commonPool-worker-3 ForkJoinPool.commonPool-worker-1 ForkJoinPool.commonPool-worker-2 ForkJoinPool.commonPool-worker-3 ForkJoinPool.commonPool-worker-1 Processed 10 tasks in 4010 millis

處理10個任務花了4秒鐘。 您會注意到，僅使用了3個ForkJoinPool線程，并且與并行流不同，沒有使用main線程。

方法4：將CompletableFutures與自定義執行器一起使用

與并行流相比， CompletableFuture的優點之一是它們允許您指定其他Executor來向其提交任務。 這意味著您可以根據應用程序選擇更合適的線程數。 由于我的示例不是很占用CPU，因此可以選擇將線程數增加到大于Runtime.getRuntime().getAvailableProcessors() ，如下所示：

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(); }

輸出是

pool-1-thread-2 pool-1-thread-4 pool-1-thread-3 pool-1-thread-1 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 1009 millis

經過改進，現在僅需1秒即可處理10個任務。

如您所見， CompletableFuture s提供了對線程池大小的更多控制，如果您的任務涉及I / O，則應使用CompletableFuture 。 但是，如果您要執行CPU密集型操作，則線程數不會超過處理器沒有意義，因此請選擇并行流，因為它更易于使用。

翻譯自: https://www.javacodegeeks.com/2016/06/java-8-completablefuture-vs-parallel-stream.html

java8并行流

總結

以上是生活随笔為你收集整理的java8并行流_Java 8：CompletableFuture与并行流的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。