CompletableFuture 无法与某些 java 8 流代码一样工作

Question

我关注this SO answer

下面的代码 sn-p 工作得很好。使用所有线程

片段 1

scrolledPage.stream()
    .filter(this::isUserDoesntHaveId)
    .map(this::processSingle)                                          // processSingle method return supplier
    .map(task -> CompletableFuture.supplyAsync(task, executorService)) // map to CompletableFuture
    .collect(Collectors.toList())                                      // Collect those as list
    .stream()                                                          // then again making another stream out of that.
    .map(CompletableFuture::join)
    .filter(Objects::nonNull)
    .collect(Collectors.toList());

我对这段代码感到困惑

    .map(task -> CompletableFuture.supplyAsync(task, executorService)) // map to CompletableFuture
    .collect(Collectors.toList())                                      // Collect those as list
    .stream()                                                          // then again making another stream out of that.
    .map(CompletableFuture::join)

为什么我需要收集然后再次制作它。我已经测试过如下。

片段 2

scrolledPage.stream()
    .filter(this::isUserDoesntHaveId)
    .map(this::processSingle)                                           // processSingle method return supplier
    .map(task -> CompletableFuture.supplyAsync(task, executorService)) // map to CompletableFuture
    .map(CompletableFuture::join)               // this code is similar without the collecting part.
    .filter(Objects::nonNull)
    .collect(Collectors.toList());

使用我看到的最后一个代码 sn-p 只使用线程池中的一个线程。但是第一个使用每个线程。这两个代码sn-ps有什么区别。

Answer 1

第一个 sn-p 可以分为两部分。 在第一部分中，您使用 CompletableFuture 将所有任务提交到异步进程，并且当它立即返回 Future 对象时，流将处理并收集列表中的所有未来。

List<Future> futures = scrolledPage.stream()
    ...
    .map(task -> CompletableFuture.supplyAsync(task, executorService))
    .collect(Collectors.toList())

稍后，流处理将在您使用join 时等待结果，但届时所有线程都将开始工作。因此它能够利用所有线程。

    futures.stream()
       .map(CompletableFuture::join)
       ...
       .collect(Collectors.toList());

Streams 进行惰性求值意味着它只会在需要时处理元素。 （需求由终端操作触发，本例为collect）。

在第二个 sn-p 中，map(CompletableFuture::join) 让线程在处理流中的下一个元素之前等待结果

.stream()
...
.map(task -> CompletableFuture.supplyAsync(task, executorService))
.map(CompletableFuture::join)
...
.collect(Collectors.toList())

因此，您在流中的任务（下一个元素）只有在第一个任务完成时才会被处理。这将使您的任务一个接一个地按顺序执行。