Java 流在完成前一个地图之前开始下一个地图

Question

我有 2 个执行者：

ExecutorService executorDownload = Executors.newFixedThreadPool(n);
ExecutorService executorCalculate = Executors.newFixedThreadPool(m);

首先我需要将任务放入 executorDownload 中，然后在它们完成后将它们放入 executorCalculate 中，然后得到结果。我写了下一个流：

long count = IntStream.range(0, TASK_NUMBER)
            .boxed()
            .parallel()
            .map(i -> executorDownload.submit(new Download(i)))
            .map(future -> calculateResultFuture(executorCalculate, future))
            .filter(Objects::nonNull)
            .filter(Main::isFutureCalculated)
            .count();

public static Future<CalculateResult> calculateResultFuture(ExecutorService executorCalculate, Future<DownloadResult> future) {
    try {
        return executorCalculate.submit(new Calculate(future.get()));
    } catch (InterruptedException | ExecutionException e) {
        e.printStackTrace();
    }
    return null;
}

public static boolean isFutureCalculated(Future<CalculateResult> future) {
    try {
        return future.get().found;
    } catch (InterruptedException | ExecutionException e) {
        e.printStackTrace();
    }
    return false;
}

有没有可能开始

.map(future -> calculateResultFuture(executorCalculate, future))

之前

.map(i -> executorDownload.submit(new Download(i)))

结束。我需要在第一张地图开始后开始第二张地图。

Answer 1

您只需要了解流会一个一个地处理元素，这意味着一个元素会在下一个元素进入第一个中间步骤之前遍历整个管道（而不是所有元素一起通过每个步骤，当然中间步骤除外需要所有元素才能完成工作）。

这意味着，在您的情况下，第一个元素的 .get() 将被阻止，从而阻止第二个元素进入第一个任务。

要强制所有元素通过第一次提交（也适用于第二次提交），您需要强制流在开始阻塞之前提交所有任务，例如：

List<Future<DownloadResult>> downloadTasks = IntStream.range(0, TASK_NUMBER)
        .mapToObj(i -> executorDownload.submit(new Download(i)))
        .collect(Collectors.toList());
        //removed .parallel()

这将强制所有异步任务启动，之后您可以对第二个异步批处理执行相同操作：

List<Future<CalculateResult>> calculateResults = downloadTasks.stream()
        .map(future -> calculateResultFuture(executorCalculate, future))
        .filter(Objects::nonNull)
        .collect(Collectors.toList());

这也将强制所有任务提交给第二个执行者。从这里，您可以.get() 无需进行不必要的等待：

long count = calculateResults.stream()
        .filter(Main::isFutureCalculated)
        .count();

---

现在，虽然这将消除批次中元素之间不必要的等待，但它仍然有可能在批次之间产生不必要的等待（如果第一个元素完成第一个任务，它将等待所有其余元素完成第一个任务，然后再继续执行第二个任务批）。为了解决这个问题，您可能需要不同的实现。这是一个为此设计的可完成期货链：

List<Completable<CalculateResult>> calculateResult = IntStream.range(0, TASK_NUMBER)
     .mapToObj(i -> CompletableFuture.supplyAsync(() -> callDownload(i), executorDownload)
             .thenApplyAsync(downloadResult -> calculateResultFuture(downloadResult), executorCalculate))
     .collect(Collectors.toList());

long count = calculateResult.stream().map(f -> isFutureCalculated(f)).count();

thenApplyAsync 将在第一个任务完成后接管第二个任务，元素方面。

当然，这需要您稍微更改 API，以便直接调用下载方法。我使用callDownload(i) 运行new Download(i).call() 运行的相同逻辑。 calculateResultFuture() 也将被更改以删除 Future 参数。

Answer 2

如果我正确理解您的问题，您需要在获得功能结果后才开始第二个 .map()。 您可以添加 .peek(Main::waitFeatureResult) 和另一种方法 waitFeatureResult:

  public static void waitFeatureResult(Future<CalculateResult> future) {
        while (future.isDone()) {
            break;
        }
    }

或编辑您的方法：

public static boolean isFutureCalculated(Future<CalculateResult> future) {
    try {
        while(future.isDone){
          return future.get().found;
        }
    } catch (InterruptedException | ExecutionException e) {
        e.printStackTrace();
        return false;
    }       
}