How Can I Use Custom Thread Pools with Java 8 Parallel Streams?

Customizable Thread Pools in Java 8 Parallel Streams

Controlling thread resource allocation within parallel streams enhances application performance and modularity. However, it remains unclear how to designate custom thread pools for specific stream operations in Java 8.

This article explores the issue and provides a solution, overcoming the limitation of shared thread pools in parallel streams.

Problem Statement

Consider a server application utilizing parallel streams. To prevent sluggish tasks within one module from impacting others, thread pool compartmentalization is desired. However, the standard parallel stream implementation employs a shared thread pool for all operations.

As exemplified by the following code snippet, a compromised task delays other threads indefinitely:

// CPU-intensive tasks in separate threads leveraging parallel streams
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));

Solution

To execute parallel operations within a designated fork-join pool, employ it as a task within the pool. This isolates the operation from the shared pool:

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

This technique leverages ForkJoinTask.fork()'s behavior: "Asynchronously execute this task in the pool the current task is running in, if applicable, or using the ForkJoinPool.commonPool() if not in ForkJoinPool()."

The above is the detailed content of How Can I Use Custom Thread Pools with Java 8 Parallel Streams?. For more information, please follow other related articles on the PHP Chinese website!