流处理线程池耗尽答案

【问题标题】：Stream processing thread pool exhaustion流处理线程池耗尽
【发布时间】：2015-02-19 11:09:50
【问题描述】：

我发现 JVM 只使用一个线程池来并行处理流，这很困难。我们在一个大型流上设置了一个 I/O 阻塞函数，这会导致与不相关的并行流一起使用的不相关或其他快速函数的活跃度问题。

流中没有允许使用备用线程池的方法。

有没有一种简单的方法可以避免这个问题，或许可以通过某种方式指定使用哪个线程池？

【问题讨论】：

Custom thread pool in Java 8 parallel stream 的可能重复项
有一个技巧——在上面链接的问题中提出了建议——但你应该记住，流主要是一种并行化计算的机制，而不是 IO，所以你在边缘操作。我们正在研究如何将其中一些用例带入主流。
@BrianGoetz 听我说，请考虑对 API 进行扩展，也许是 parallel() 方法的一个参数，它允许将单独的线程池用于并行流处理 - 一些计算也很重，可能会导致活力问题。虽然我看到“非常长”的时间（例如 IO）超出了设计的范围，但使用流来驱动这种使用是很自然的，恕我直言，应该有一种简单的方法来减轻这种使用的影响.对我来说，它打破了 Josh 的“最少惊讶”规则。圣诞快乐:)
@Bohemian 听到您不是第一个提出这个建议的人可能不会感到惊讶。但是，我们有充分的理由不做出这个“显而易见”的选择，我仍然相信我们做出了正确的决定。

标签： parallel-processing java-8 threadpool java-stream

【解决方案1】：

我编写了一个名为StreamEx 的小型库，它可以向自定义 FJP 提交任务。所以你可以写

ForkJoinPool forkJoinPool = new ForkJoinPool(2);
int[] primes = IntStreamEx.range(1, 1_000_000)
    .parallel(forkJoinPool)
    .filter(PrimesPrint::isPrime).toArray();

它只是记住您的池并在其中启动终端操作并加入结果。只是上述解决方案的语法糖。

【讨论】：

【解决方案2】：

您可以将阻塞操作包装在 ForkJoinPool.ManagedBlocker 中，如下所示：

static <T> Supplier<T> blocking(Supplier<T> supplier) {
    return new Supplier<T>() {
        volatile T result;

        @Override
        public T get() {
            try {
                ForkJoinPool.managedBlock(new ManagedBlocker() {
                    @Override
                    public boolean block() {
                        result = supplier.get();
                        return true;
                    }

                    @Override
                    public boolean isReleasable() {
                        return result != null;
                    }
                });
            }
            catch (InterruptedException e) {
                throw new RuntimeException(e);
            }

            return result;
        }
    };
}

然后使用它，例如：

Stream.generate(blocking(() -> ...))
      .parallel()
      ...
      .collect(...);

更多信息可以在这篇博文中找到： http://zeroturnaround.com/rebellabs/java-parallel-streams-are-bad-for-your-health/

jOOλ 通过org.jooq.lambda.Blocking 为所有Java 8 FunctionalInterface 类型（如上）提供包装器，因此您可以编写：

Stream.generate(Blocking.supplier(() -> ...))
      .parallel()
      ...
      .collect(...);

或者，例如当过滤器阻塞时：

Stream....
      .parallel()
      .filter(Blocking.predicate(t -> blockingTest(t)))
      .collect(...);

（免责声明，我在 jOOλ 背后的公司工作）。

【讨论】：

IIRC 您声称 jOOλ 不关心并行处理 :-) 实际上您的解决方案（以及我的解决方案）还有另一个问题。 Stream API 创建的子任务数量通常是 4*处理器数量（对于无限流可能更多）。如果我们创建具有 100 个线程的自定义池或使用您的方法将任务标记为阻塞，则此数字不会改变。对于阻塞 IO，实际上最好使用其他东西（例如 CompletableFutures）。
@TagirValeev：jOOλ 的目的是“修复”整个 JDK（例如，还提供抛出检查异常的功能接口）。 jOOλ 的Seq 的目的是“修复”Stream 以供顺序（和有序）使用。 CompletableFuture 也将任务发送到ForkJoinPool，因此您可能会耗尽相同的线程。我同意这里的这种用法可能不是人们应该做的。但话又说回来，什么是阻塞的（在这种情况下）？是 I/O 吗？在1_000_000_000_000 之前是一个循环吗？这是一个愚蠢的指数算法吗？
OP 明确表示它是 I/O。使用 CF，您可以创建任意数量的任务，但使用并行流您无法控制它。在 4 核机器上，通常会创建多达 16 个任务，如果其中 15 个在 I/O 操作中被阻塞，那么即使使用您的技巧，也只有一个任务会继续工作。 Stream API 将不再划分输入，也不会为您创建更多任务。

【解决方案3】：

这可能类似于Custom thread pool in Java 8 parallel stream

该问题在this blog 中进一步讨论。

ForkJoinPool forkJoinPool = new ForkJoinPool(2);
forkJoinPool.submit(() ->
    //parallel task here, for example
    range(1, 1_000_000).parallel().filter(PrimesPrint::isPrime).collect(toList())
).get();

【讨论】：