Java 8 并行流中的自定义线程池

concurrency java parallel-processing java-8 java-stream

2022-08-31 04:29:22

是否可以为 Java 8 并行流指定定制线程池？我在任何地方都找不到它。

想象一下，我有一个服务器应用程序，我想使用并行流。但是应用程序很大并且是多线程的，所以我想把它分隔开来。我不希望在应用程序阻止任务的一个模块中运行缓慢的任务来自另一个模块。

如果我不能为不同的模块使用不同的线程池，这意味着我不能在大多数现实世界的情况下安全地使用并行流。

请尝试以下示例。有一些 CPU 密集型任务在单独的线程中执行。这些任务利用并行流。第一个任务已中断，因此每个步骤需要 1 秒（通过线程休眠模拟）。问题是其他线程卡住并等待损坏的任务完成。这是人为的示例，但想象一下一个 servlet 应用程序和某人向共享分叉加入池提交一个长时间运行的任务。

public class ParallelTest {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService es = Executors.newCachedThreadPool();

        es.execute(() -> runTask(1000)); //incorrect task
        es.execute(() -> runTask(0));
        es.execute(() -> runTask(0));
        es.execute(() -> runTask(0));
        es.execute(() -> runTask(0));
        es.execute(() -> runTask(0));


        es.shutdown();
        es.awaitTermination(60, TimeUnit.SECONDS);
    }

    private static void runTask(int delay) {
        range(1, 1_000_000).parallel().filter(ParallelTest::isPrime).peek(i -> Utils.sleep(delay)).max()
                .ifPresent(max -> System.out.println(Thread.currentThread() + " " + max));
    }

    public static boolean isPrime(long n) {
        return n > 1 && rangeClosed(2, (long) sqrt(n)).noneMatch(divisor -> n % divisor == 0);
    }
}

答案 1

实际上有一个技巧如何在特定的分叉连接池中执行并行操作。如果将其作为分叉连接池中的任务执行，它将保留在那里并且不使用公共任务。

final int parallelism = 4;
ForkJoinPool forkJoinPool = null;
try {
    forkJoinPool = new ForkJoinPool(parallelism);
    final List<Integer> primes = forkJoinPool.submit(() ->
        // Parallel task here, for example
        IntStream.range(1, 1_000_000).parallel()
                .filter(PrimesPrint::isPrime)
                .boxed().collect(Collectors.toList())
    ).get();
    System.out.println(primes);
} catch (InterruptedException | ExecutionException e) {
    throw new RuntimeException(e);
} finally {
    if (forkJoinPool != null) {
        forkJoinPool.shutdown();
    }
}

该技巧基于ForkJoinTask.fork，它指定：“安排在当前任务正在运行的池中异步执行此任务（如果适用），或者使用如果不是ForkJoinPool.commonPool()inForkJoinPool()"

答案 2

并行流使用默认值，默认情况下，该默认值具有一个线程，因为您有处理器，如返回的那样（这意味着并行流为调用线程保留一个处理器）。ForkJoinPool.commonPoolRuntime.getRuntime().availableProcessors()

对于需要单独或自定义池的应用程序，可以使用给定的目标并行度级别构造ForkJoinPool;默认情况下，等于可用处理器的数量。

这也意味着，如果嵌套的并行流或同时启动的多个并行流，它们将共享同一池。优点：您永远不会使用超过默认值（可用处理器的数量）。缺点：您可能无法将“所有处理器”分配给您启动的每个并行流（如果您碰巧有多个并行流）。（显然，您可以使用托管阻止程序来规避这种情况。

要更改并行流的执行方式，您可以

将并行流执行提交到您自己的 ForkJoinPool：或yourFJP.submit(() -> stream.parallel().forEach(soSomething)).get();
您可以使用系统属性更改公用池的大小：对于 20 个线程的目标并行度。System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20")

后者在我的机器上有8个处理器的例子。如果我运行以下程序：

long start = System.currentTimeMillis();
IntStream s = IntStream.range(0, 20);
//System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20");
s.parallel().forEach(i -> {
    try { Thread.sleep(100); } catch (Exception ignore) {}
    System.out.print((System.currentTimeMillis() - start) + " ");
});

输出为：

215 216 216 216 216 216 216 216 315 316 316 316 316 316 316 316 415 416 416 416

因此，您可以看到并行流一次处理 8 个项目，即它使用 8 个线程。但是，如果我取消注释的行，则输出为：

215 215 215 215 215 216 216 216 216 216 216 216 216 216 216 216 216 216 216 216

这一次，并行流使用了 20 个线程，并且流中的所有 20 个元素都已同时处理。