强制 Java 流提前执行部分管道以将阻塞任务提交到线程池

Question

我有一个我想要处理的对象列表，Java8 流 API 看起来是最简洁易读的方式。

但是我需要对这些对象执行的一些操作包括阻塞 IO（例如读取数据库）——所以我想将这些操作提交到具有几十个线程的线程池。

一开始我想做一些类似的事情：

myObjectList
    .stream()
    .filter(wrapPredicate(obj -> threadPoolExecutor.submit(
            () -> longQuery(obj)          // returns boolean
    ).get())                              // wait for future & unwrap boolean
    .map(filtered -> threadPoolExecutor.submit(
            () -> anotherQuery(filtered)  // returns Optional
    ))
    .map(wrapFunction(Future::get))
    .filter(Optional::isPresent)
    .map(Optional::get)
    .collect(toList());

wrapPredicate 和 wrapFunction 仅用于检查异常重新抛出。

但是，显然，对Future.get() 的调用将阻塞流的线程，直到对给定对象的查询完成，并且在此之前流不会继续。所以一次只处理一个对象，线程池没有意义。

我可以使用并行流，但我需要希望默认的 ForkJoinPool 足以满足此要求。或者只是增加"java.util.concurrent.ForkJoinPool.common.parallelism"，但我不想为了那个流而改变整个应用程序的设置。我可以在自定义ForkJoinPool 中创建流，但我看到it does not guarantee that level of parallelism。

所以我最终得到了类似的东西，只是为了保证在等待期货完成之前将所有需要的任务提交到线程池：

myObjectList
    .stream()
    .map(obj -> Pair.of(obj, threadPoolExecutor.submit(
                    () -> longQuery(obj)             // returns boolean
        ))
    )
    .collect(toList()).stream()                      // terminate stream to actually submit tasks to the pool
    .filter(wrapPredicate(p -> p.getRight().get()))  // wait & unwrap future after all tasks are submitted
    .map(Pair::getLeft)
    .map(filtered -> threadPoolExecutor.submit(
            () -> anotherQuery(filtered)             // returns Optional
    ))
    .collect(toList()).stream()                      // terminate stream to actually submit tasks to the pool
    .map(wrapFunction(Future::get))                  // wait & unwrap futures after all submitted
    .filter(Optional::isPresent)
    .map(Optional::get)
    .collect(toList());

有没有明显更好的方法来实现这一点？

一种更优雅的方式来告诉流“现在对流中的每个对象执行流水线步骤”，然后继续处理.collect(toList()).stream() 以外的其他处理，以及过滤Future 效果的更好方法比将其打包到 Apache Commons Pair 以便稍后在 Pair::getRight 上过滤？或者也许是一个完全不同的方法来解决这个问题？

Answer 1

您可以通过使用大大简化您的代码

myObjectList.stream()
    .map(obj -> threadPoolExecutor.submit(
                    () -> longQuery(obj)? anotherQuery(obj).orElse(null): null))
    .collect(toList()).stream()
    .map(wrapFunction(Future::get))
    .filter(Objects::nonNull)
    .collect(toList());

有一点是，如果您稍后将anotherQuery 提交给同一个执行者，并发性不会有任何改善。因此，您可以在longQuery 返回true 之后直接执行它。此时，obj 仍在范围内，因此您可以将其用于anotherQuery。

通过提取Optional的结果，使用null作为失败的表示，我们可以获得相同的缺失结果表示，因为longQuery返回false或anotherQuery返回一个空Optional。所以在提取Future的结果后，我们要做的就是到.filter(Objects::nonNull)。

你必须先提交作业，收集Futures，然后才能得到实际结果的逻辑没有改变。无论如何也没有办法。其他便利方法或框架所能提供的只是隐藏这些对象的实际临时存储。

Answer 2

我认为主要问题的答案是否定的。为了“执行”一个流，你需要一个终端操作。但可能还有改进的余地。

您至少可以通过收集到地图而不是列表来摆脱这对：

stream.collect(toMap(Function.identity(),
                     obj -> threadPoolExecutor.submit(() -> longQuery(obj))))
      .entrySet()
      .stream()
      .filter(wrapPredicate(entry -> entry.getValue().get()))
      .map(Entry::getKey)
      ...

请注意，这仅在处理的对象都不等于另一个时才有效。它使代码略短且更易于阅读，因为您不必自己创建对/条目。

Answer 3

您可以为 Java 8 并行流指定线程池。您不必更改应用程序设置。更多信息：https://stackoverflow.com/a/22269778/7123191。

Answer 4

您至少可以通过收集到地图而不是列表来摆脱这对：

stream.collect(toMap(Function.identity(),
                     obj -> threadPoolExecutor.submit(() -> longQuery(obj))))
      .entrySet()
      .stream()
      .filter(wrapPredicate(entry -> entry.getValue().get()))
      .map(Entry::getKey)
      ...