java - 在 Java8 parallelStream() 中使用 I/O + ManagedBlocker 有什么问题吗？

Question

Java 8 中默认的“paralellStream()”使用公共(public)的ForkJoinPool，如果在提交任务时公共(public)的Pool 线程耗尽，这可能是一个延迟问题。然而，在许多情况下，有足够的 CPU 能力可用并且任务足够短，因此这不是问题。如果我们确实有一些长时间运行的任务，这当然需要仔细考虑，但对于这个问题，我们假设这不是问题所在。

但是，即使有足够的 CPU 能力可用，用实际上不执行任何 CPU 绑定(bind)工作的 I/O 任务填充 ForkJoinPool 也是一种引入瓶颈的方法。 I understood that .然而，这正是我们拥有 ManagedBlocker 的目的。因此，如果我们有一个 I/O 任务，我们应该简单地允许 ForkJoinPool 在 ManagedBlocker 中管理它。这听起来非常容易。然而，令我惊讶的是，使用 ManagedBlocker 是一个相当复杂的 API，尽管它很简单。毕竟我认为这是一个普遍的问题。所以我只是构建了一个简单的实用方法，使 ManagedBlocker 易于在常见情况下使用:

public class BlockingTasks {

    public static<T> T callInManagedBlock(final Supplier<T> supplier) {
        final SupplierManagedBlock<T> managedBlock = new SupplierManagedBlock<>(supplier);
        try {
            ForkJoinPool.managedBlock(managedBlock);
        } catch (InterruptedException e) {
            throw new Error(e);
        }
        return managedBlock.getResult();
    }

    private static class SupplierManagedBlock<T> implements ForkJoinPool.ManagedBlocker {
        private final Supplier<T> supplier;
        private T result;
        private boolean done = false;

        private SupplierManagedBlock(final Supplier<T> supplier) {
            this.supplier = supplier;
        }

        @Override
        public boolean block() {
            result = supplier.get();
            done = true;
            return true;
        }

        @Override
        public boolean isReleasable() {
            return done;
        }

        public T getResult() {
            return result;
        }
    }
}

现在，如果我想并行下载几个网站的 html 代码，我可以这样下载，而不会造成任何 I/O 问题:

public static void main(String[] args) {
    final List<String> pagesHtml = Stream
        .of("https://google.com", "https://stackoverflow.com", "...")
        .map((url) -> BlockingTasks.callInManagedBlock(() -> download(url)))
        .collect(Collectors.toList());
}

Java 没有像上面的 BlockingTasks 这样的类(或者我没有找到它？)，这让我有点惊讶，但它并不难构建。

当我在谷歌上搜索“java 8 parallel stream”时，我在前四个结果中得到了那些声称由于 I/O 问题 Fork/Join 在 Java 中很糟糕的文章:

• https://dzone.com/articles/think-twice-using-java-8
• http://zeroturnaround.com/rebellabs/java-parallel-streams-are-bad-for-your-health/ (至少提到 ManagedBlocker 但也说“在一个不同的用例中你可以给它一个 ManagedBlocker 实例”。它没有提到为什么不在这种情况下.

我对我的搜索词做了一些修改，虽然有很多人提示生活多么可怕，但我发现没有人在谈论像上面这样的解决方案。因为我不像 Marvin(大脑像行星)而且 Java 8 已经可用了很长一段时间，所以我怀疑我在那里提出的建议存在严重错误。

我做了一个小测试:

public static void main(String[] args) {
    System.out.println(DateTimeFormatter.ISO_LOCAL_TIME.format(LocalTime.now()) + ": Start");
    IntStream.range(0, 10).parallel().forEach((x) -> sleep());
    System.out.println(DateTimeFormatter.ISO_LOCAL_TIME.format(LocalTime.now()) + ": End");
}

public static void sleep() {
    try {
        System.out.println(DateTimeFormatter.ISO_LOCAL_TIME.format(LocalTime.now()) + ": Sleeping " + Thread.currentThread().getName());
        Thread.sleep(10000);
    } catch (InterruptedException e) {
        throw new Error(e);
    }
}

我运行了它并得到了以下结果:

18:41:29.021: Start
18:41:29.033: Sleeping main
18:41:29.034: Sleeping ForkJoinPool.commonPool-worker-1
18:41:29.034: Sleeping ForkJoinPool.commonPool-worker-2
18:41:29.034: Sleeping ForkJoinPool.commonPool-worker-5
18:41:29.034: Sleeping ForkJoinPool.commonPool-worker-4
18:41:29.035: Sleeping ForkJoinPool.commonPool-worker-6
18:41:29.035: Sleeping ForkJoinPool.commonPool-worker-3
18:41:29.035: Sleeping ForkJoinPool.commonPool-worker-7
18:41:39.034: Sleeping main
18:41:39.034: Sleeping ForkJoinPool.commonPool-worker-1
18:41:49.035: End

所以在我的 8 CPU 计算机上，ForkJoinPool 自然选择 8 个线程，完成前 8 个任务，最后完成最后两个任务，这意味着这需要 20 秒，如果有其他任务在池中排队可能仍然没有使用明显空闲的 CPU(除了最近 10 秒的 6 个内核)。

然后我用...

IntStream.range(0, 10).parallel().forEach((x) -> callInManagedBlock(() -> { sleep(); return null; }));

...而不是...

IntStream.range(0, 10).parallel().forEach((x) -> sleep());

...得到如下结果:

18:44:10.93: Start
18:44:10.945: Sleeping main
18:44:10.953: Sleeping ForkJoinPool.commonPool-worker-7
18:44:10.953: Sleeping ForkJoinPool.commonPool-worker-1
18:44:10.953: Sleeping ForkJoinPool.commonPool-worker-6
18:44:10.953: Sleeping ForkJoinPool.commonPool-worker-3
18:44:10.955: Sleeping ForkJoinPool.commonPool-worker-2
18:44:10.956: Sleeping ForkJoinPool.commonPool-worker-4
18:44:10.956: Sleeping ForkJoinPool.commonPool-worker-5
18:44:10.956: Sleeping ForkJoinPool.commonPool-worker-0
18:44:10.956: Sleeping ForkJoinPool.commonPool-worker-11
18:44:20.957: End

在我看来这很有效，启动了额外的线程来补偿我的模拟“阻塞 I/O 操作”( sleep )。时间缩短到 10 秒，我想如果我将更多任务排队，这些任务仍然可以使用可用的 CPU 能力。

如果 I/O 操作包装在 ManagedBlock 中，这个解决方案或通常在流中使用 I/O 有什么问题吗？

Answer 1

简而言之，是的，您的解决方案存在一些问题。在并行流中使用阻塞代码肯定会有所改进，一些第三方库提供了类似的解决方案(例如，参见 jOOλ 库中的 Blocking 类)。但是这个解决方案并没有改变 Stream API 中使用的内部拆分策略。 Stream API 创建的子任务数量由 AbstractTask 类中的预定义常量控制:

/**
 * Default target factor of leaf tasks for parallel decomposition.
 * To allow load balancing, we over-partition, currently to approximately
 * four tasks per processor, which enables others to help out
 * if leaf tasks are uneven or some processors are otherwise busy.
 */
static final int LEAF_TARGET = ForkJoinPool.getCommonPoolParallelism() << 2;

如您所见，它比普通池并行度(默认情况下的 CPU 核心数)大四倍。真正的拆分算法有点棘手，但粗略地说，即使所有任务都处于阻塞状态，您也不能拥有超过 4x-8x 的任务。

例如，如果您有 8 个 CPU 内核，您的 Thread.sleep() 测试将很好地运行到 IntStream.range(0, 32)(如 32 = 8 * 4)。但是，对于 IntStream.range(0, 64)，您将有 32 个并行任务，每个任务处理两个输入数字，因此整个处理将花费 20 秒，而不是 10 秒。