理解链式可完成期货的并行执行

Question

我有一个关于Java流和链式可完成期货如何执行的问题。

我的问题是：如果我运行下面的代码，调用执行（），列表中有10个项目需要大约11秒才能完成（列表中的项目数加1）。这是因为我有两个线程并行工作：第一个执行digItUp操作，一旦完成，第二个执行fillItBackIn操作，第一个开始处理digItUp列表中的下一个项目。

如果我注释掉第36行（.collect（Collectors.toList（））），那么execute（）方法需要大约20秒才能完成。Thread不平行运行；对于列表中的每个项目，digItUp操作完成，然后在处理列表中的下一个项目之前，fillItBackIn操作按顺序完成。

我不清楚为什么排除（。收集（Collectors.toList（）））会改变这种行为。有人能解释一下吗？

完整课程：

package com.test;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

public class SimpleExample {

    private final ExecutorService diggingThreadPool = Executors.newFixedThreadPool(1);
    private final ExecutorService fillingThreadPool = Executors.newFixedThreadPool(1);

    public SimpleExample() {

    }

    public static void main(String[] args) {
        List<Double> holesToDig = new ArrayList<>();
        Random random = new Random();
        for (int c = 0; c < 10; c++) {
            holesToDig.add(random.nextDouble(1000));
        }
        new SimpleExample().execute(holesToDig);
    }

    public void execute(List<Double> holeVolumes) {
        long start = System.currentTimeMillis();
        holeVolumes.stream()
                .map(volume -> {
                    CompletableFuture<Double> digItUpCF = CompletableFuture.supplyAsync(() -> digItUp(volume), diggingThreadPool);
                    return digItUpCF.thenApplyAsync(volumeDugUp -> fillItBackIn(volumeDugUp), fillingThreadPool);
                })
                .collect(Collectors.toList())
                .forEach(cf -> {
                    Double volume = cf.join();
                    System.out.println("Dug a hole and filled it back in.  Net volume: " + volume);
                });
        System.out.println("Dug up and filled back in " + holeVolumes.size() + " holes in " + (System.currentTimeMillis() - start) + " ms");
    }

    public Double digItUp(Double volume) {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
        }
        System.out.println("Dug hole with volume " + volume);
        return volume;
    }

    public Double fillItBackIn(Double volumeDugUp) {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
        }
        System.out.println("Filled back in hole of volume " + volumeDugUp);
        return 0.0;
    }
}

Answer 1

原因是collect（Collectors.toList（））是一个终端操作，因此它会触发流管道（请记住，流是惰性计算的）。因此，当您调用collect时，所有CompletableFuture实例都会被构造并放置在列表中。这意味着有一个CompletableFuture链，其中每一个都是由两个阶段组成的链，我们称之为X和Y。

每当第一个线程执行器完成一个X阶段时，它就可以自由地处理下一个组合的CompletableFuture的X阶段，而另一个线程执行器则处理上一个CompletableFuture的阶段Y。这是我们直觉预期的结果。

另一方面，当您不调用collect时，则在本例中，forEach是终端操作。然而，在这种情况下，流中的每个元素都是按顺序处理的（以确认是否尝试切换到parallelStream（）），因此对于第一个CompletableFuture，会执行阶段X和Y。只有当前一个流元素的阶段Y完成时，forEach才会移动到流管道中的第二个元素，只有到那时，新的CompletableFuture才会从原始的Double值映射。