Java8并行流中的自定义线程池

Question

是否可以为Java8并行流指定自定义线程池？我到处都找不到它。

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

请尝试以下示例。有些CPU密集型任务在单独的线程中执行。这些任务利用并行流。第一个任务被打破，因此每一步需要1秒（通过线程Hibernate模拟）。问题是其他线程会被卡住，等待中断的任务完成。这是一个虚构的示例，但假设一个servlet应用程序和一个人将一个长期运行的任务提交给共享的fork连接池。

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);
    }
}

Answer 1

实际上，在特定的fork-join池中执行并行操作有一个诀窍。如果您在fork-join池中作为任务执行它，则它将保留在那里，而不使用通用的。

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();
    }
}

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