强制Java stream提前执行管道的一部分，以便将阻塞任务提交到线程池

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中创建流，但我发现它不能保证这种级别的并行性。

所以我以这样的方式结束，只是为了保证在等待futures完成之前，所有需要的任务都提交到线程池：

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 Commonspair中，以便稍后对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)。

在获得实际结果之前，您必须首先提交作业（收集future)的逻辑不会改变。反正也没办法绕过。其他方法或框架所能提供的所有便利性都是隐藏这些对象的实际临时存储。