Custom thread pool in Java 8 parallel stream

Question

Is it possible to specify a custom thread pool for Java 8 parallel stream  I can not find it anywhere    Imagine that I have a server application and I would like to use parallel streams  But the application is large and multi-threaded so I want to compartmentalize it  I do not want a slow running task in one module of the applicationblock tasks from another module   If I can not use different thread pools for different modules  it means I can not safely use parallel streams in most of real world situations   Try the following example  There are some CPU intensive tasks executed in separate threads  The tasks leverage parallel streams  The first task is broken  so each step takes 1 second  simulated by thread sleep   The issue is that other threads get stuck and wait for the broken task to finish  This is contrived example  but imagine a servlet app and someone submitting a long running task to the shared fork join pool    public class ParallelTest       public static void main String   args  throws InterruptedException           ExecutorService es   Executors newCachedThreadPool             es execute    - gt  runTask 1000      incorrect task         es execute    - gt  runTask 0            es execute    - gt  runTask 0            es execute    - gt  runTask 0            es execute    - gt  runTask 0            es execute    - gt  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 - gt  Utils sleep delay   max                    ifPresent max - gt  System out println Thread currentThread           max               public static boolean isPrime long n            return n  gt  1  amp  amp  rangeClosed 2   long  sqrt n   noneMatch divisor - gt  n   divisor    0

User · Accepted Answer

There actually is a trick how to execute a parallel operation in a specific fork-join pool. If you execute it as a task in a fork-join pool, it stays there and does not use the common one.

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

The trick is based on ForkJoinTask.fork which specifies: "Arranges to asynchronously execute this task in the pool the current task is running in, if applicable, or using the ForkJoinPool.commonPool() if not inForkJoinPool()"

User · Answer

To measure the actual number of used threads  you can check Thread activeCount         Runnable r      - gt  IntStream              range -42   42               parallel                map i - gt  Thread activeCount                 max                ifPresent System out  println        ForkJoinPool commonPool   submit r  join        new ForkJoinPool 42  submit r  join      This can produce on a 4-core CPU an output like   5    common pool 23    custom pool   Without  parallel   it gives   3    common pool 4    custom pool

User · Answer

Here is how I set the max thread count flag mentioned above programatically and a code sniped to verify that the parameter is honored System setProperty  quot java util concurrent ForkJoinPool common parallelism quot    quot 2 quot    Set lt String gt  threadNames   Stream iterate 0  n - gt  n   1     parallel      limit 100000     map i - gt  Thread currentThread   getName       collect Collectors toSet     System out println threadNames       Output - gt   ForkJoinPool commonPool-worker-1  Test worker  ForkJoinPool commonPool-worker-3

User · Answer

The parallel streams use the default ForkJoinPool commonPool which by default has one less threads as you have processors  as returned by Runtime getRuntime   availableProcessors    This means that parallel streams leave one processor for the calling thread    For applications that require separate or custom pools  a ForkJoinPool may be constructed with a given target parallelism level  by default  equal to the number of available processors   This also means if you have nested parallel streams or multiple parallel streams started concurrently  they will all share the same pool   Advantage  you will never use more than the default  number of available processors    Disadvantage  you may not get  quot all the processors quot  assigned to each parallel stream you initiate  if you happen to have more than one     Apparently you can use a ManagedBlocker to circumvent that   To change the way parallel streams are executed  you can either  submit the parallel stream execution to your own ForkJoinPool  yourFJP submit    - gt  stream parallel   forEach soSomething   get    or you can change the size of the common pool using system properties  System setProperty  quot java util concurrent ForkJoinPool common parallelism quot    quot 20 quot   for a target parallelism of 20 threads  However  this no longer works after the backported patch https   bugs openjdk java net browse JDK-8190974    Example of the latter on my machine which has 8 processors  If I run the following program  long start   System currentTimeMillis    IntStream s   IntStream range 0  20     System setProperty  quot java util concurrent ForkJoinPool common parallelism quot    quot 20 quot    s parallel   forEach i - gt        try   Thread sleep 100     catch  Exception ignore         System out print  System currentTimeMillis   - start     quot   quot         The output is   215 216 216 216 216 216 216 216 315 316 316 316 316 316 316 316 415 416 416 416  So you can see that the parallel stream processes 8 items at a time  i e  it uses 8 threads  However  if I uncomment the commented line  the output is   215 215 215 215 215 216 216 216 216 216 216 216 216 216 216 216 216 216 216 216  This time  the parallel stream has used 20 threads and all 20 elements in the stream have been processed concurrently

User · Answer

Until now  I used the solutions described in the answers of this question  Now  I came up with a little library called Parallel Stream Support for that   ForkJoinPool pool   new ForkJoinPool NR OF THREADS   ParallelIntStreamSupport range 1  1 000 000  pool       filter PrimesPrint  isPrime       collect toList      But as  PabloMatiasGomez pointed out in the comments  there are drawbacks regarding the splitting mechanism of parallel streams which depends heavily on the size of the common pool  See Parallel stream from a HashSet doesn  39 t run in parallel    I am using this solution only to have separate pools for different types of work but I can not set the size of the common pool to 1 even if I don t use it

User · Answer

We can change the default parallelism using the following property   -Djava util concurrent ForkJoinPool common parallelism 16   which can set up to use more parallelism

User · Answer

Alternatively to the trick of triggering the parallel computation inside your own forkJoinPool you can also pass that pool to the CompletableFuture supplyAsync method like in   ForkJoinPool forkJoinPool   new ForkJoinPool 2   CompletableFuture lt List lt Integer gt  gt  primes   CompletableFuture supplyAsync    - gt        parallel task here  for example     range 1  1 000 000  parallel   filter PrimesPrint  isPrime  collect toList          forkJoinPool

User · Answer

Go to get AbacusUtil  Thread number can by specified for parallel stream  Here is the sample code   LongStream range 4  1 000 000  parallel threadNum       Disclosure  I m the developer of AbacusUtil

User · Answer

I tried the custom ForkJoinPool as follows to adjust the pool size   private static Set lt String gt  ThreadNameSet   new HashSet lt  gt     private static Callable lt Long gt  getSum         List lt Long gt  aList   LongStream rangeClosed 0  10 000 000  boxed   collect Collectors toList         return    - gt  aList parallelStream                peek  i  - gt                    String threadName   Thread currentThread   getName                    ThreadNameSet add threadName                               reduce 0L  Long  sum      private static void testForkJoinPool         final int parallelism   10       ForkJoinPool forkJoinPool   null      Long result   0L      try           forkJoinPool   new ForkJoinPool parallelism           result   forkJoinPool submit getSum    get      this makes it an overall blocking call        catch  InterruptedException   ExecutionException e            e printStackTrace          finally           if  forkJoinPool    null                forkJoinPool shutdown      always remember to shutdown the pool                     out println result       out println ThreadNameSet       Here is the output saying the pool is using more threads than the default 4    50000005000000  ForkJoinPool-1-worker-8  ForkJoinPool-1-worker-9  ForkJoinPool-1-worker-6  ForkJoinPool-1-worker-11  ForkJoinPool-1-worker-10  ForkJoinPool-1-worker-1  ForkJoinPool-1-worker-15  ForkJoinPool-1-worker-13  ForkJoinPool-1-worker-4  ForkJoinPool-1-worker-2    But actually there is a weirdo  when I tried to achieve the same result using ThreadPoolExecutor as follows   BlockingDeque blockingDeque   new LinkedBlockingDeque 1000   ThreadPoolExecutor fixedSizePool   new ThreadPoolExecutor 10  20  60  TimeUnit SECONDS  blockingDeque  new MyThreadFactory  my-thread       but I failed    It will only start the parallelStream in a new thread and then everything else is just the same  which again proves that the parallelStream will use the ForkJoinPool to start its child threads

User · Answer

If you don t want to rely on implementation hacks  there s always a way to achieve the same by implementing custom collectors that will combine map and collect semantics    and you wouldn t be limited to ForkJoinPool  list stream      collect parallel i - gt  process i   executor  4      join    Luckily  it s done already here and available on Maven Central  http   github com pivovarit parallel-collectors Disclaimer  I wrote it and take responsibility for it

User · Answer

Note   There appears to be a fix implemented in JDK 10 that ensures the Custom Thread Pool uses the expected number of threads   Parallel stream execution within a custom ForkJoinPool should obey the parallelism https   bugs openjdk java net browse JDK-8190974

User · Answer

you can try implementing this ForkJoinWorkerThreadFactory and inject it to Fork-Join class   public ForkJoinPool int parallelism                          ForkJoinWorkerThreadFactory factory                          UncaughtExceptionHandler handler                          boolean asyncMode            this checkParallelism parallelism                checkFactory factory                handler               asyncMode   FIFO QUEUE   LIFO QUEUE                ForkJoinPool-    nextPoolId      -worker-            checkPermission            you can use this constructor of Fork-Join pool to do this   notes --  1  if you use this  take into consideration that based on your implementation of new threads  scheduling from JVM will be affected  which generally schedules fork-join threads to different cores treated as a computational thread    2  task scheduling by fork-join to threads won t get affected   3  Haven t really figured out how parallel stream is picking threads    from fork-join couldn t find proper documentation on it   so try    using a different threadNaming factory so as to make sure  if threads    in parallel stream are being picked from customThreadFactory that you    provide   4  commonThreadPool won t use this customThreadFactory

User · Answer

If you don t mind using a third-party library  with cyclops-react you can mix sequential and parallel Streams within the same pipeline and provide custom ForkJoinPools  For example   ReactiveSeq range 1  1 000 000               foldParallel new ForkJoinPool 10                             s- gt s filter i- gt true                                 peek i- gt System out println  Thread     Thread currentThread   getId                                    max Comparator naturalOrder        Or if we wished to continue processing within a sequential Stream   ReactiveSeq range 1  1 000 000               parallel new ForkJoinPool 10                         s- gt s filter i- gt true                             peek i- gt System out println  Thread     Thread currentThread   getId                   map this  processSequentially               forEach System out  println      Disclosure I am the lead developer of cyclops-react

User · Answer

The original solution  setting the ForkJoinPool common parallelism property  no longer works  Looking at the links in the original answer  an update which breaks this has been back ported to Java 8  As mentioned in the linked threads  this solution was not guaranteed to work forever  Based on that  the solution is the forkjoinpool submit with  get solution discussed in the accepted answer  I think the backport fixes the unreliability of this solution also   ForkJoinPool fjpool   new ForkJoinPool 10   System out println  stream parallel    IntStream range   IntStream range 0  20   fjpool submit    - gt  range parallel            forEach  int theInt  - gt                        try   Thread sleep 100     catch  Exception ignore                 System out println Thread currentThread   getName       --     theInt               get    System out println  list parallelStream    int    array   IntStream range 0  20  toArray    List lt Integer gt  list   new ArrayList lt  gt     for  int theInt  array        list add theInt     fjpool submit    - gt  list parallelStream            forEach  theInt  - gt                        try   Thread sleep 100     catch  Exception ignore                 System out println Thread currentThread   getName       --     theInt               get

User · Answer

If you don t need a custom ThreadPool but you rather want to limit the number of concurrent tasks  you can use   List lt Path gt  paths   List of   path file1 csv     path file2 csv     path file3 csv   stream   map e - gt  Paths get e   collect toList     List lt List lt Path gt  gt  partitions   Lists partition paths  4      Guava method  partitions forEach group - gt  group parallelStream   forEach csvFilePath - gt              do your processing            Duplicate question asking for this is locked  so please bear me here

[java] Custom thread pool in Java 8 parallel stream

Examples related to java

Examples related to concurrency

Examples related to parallel-processing

Examples related to java-8

Examples related to java-stream