java 8 parallel stream performance

Java 8 parallel streaming internally use Fork Join Pool to break raw data and return as a stream.

So first let us discuss on Fork and Join Pool in Java 8.

Use the Parallel Fork/Join Framework

The Fork/Join framework in the java.util.concurrent package helps simplify writing parallelized code.

Before we look at the ForkJoinPool I want to explain how the fork and join principle works in general just a overview.

The fork and join principle consists of two steps which are performed recursively. These two steps are the fork step and the join step.

Fork

Join

When a task has split itself up into subtasks, the task waits until the subtasks have finished executing.

Once the subtasks have finished executing, the task may join (merge) all the results into one result.

A task that uses the fork and join principle can fork (split) itself into smaller subtasks which can be executed concurrently. This is illustrated in the complete diagram below:

The framework is an implementation of the ExecutorService interface and provides an easy-to-use concurrent platform in order to exploit multiple processors. This framework is very useful for modeling divide-and-conquer problems. Divide-and-conquer is a naturally parallel algorithmic technique. Most often we can solve the sub instances in parallel. This can lead to significant amount of parallelism since at each level of can create more instances to solve in parallel. Even if we only divide our instance into two sub instances, each of those sub instances will themselves generate two more sub-instances, and this repeats.

This approach is suitable for tasks that can be divided recursively and computed on a smaller scale; the computed results are then combined. Dividing the task into smaller tasks is forking, and merging the results from the smaller tasks is joining.

The Fork/Join framework uses the work-stealing algorithm: when a worker thread completes its work and is free, it takes (or “steals”) work from other threads that are still busy doing some work. Initially, it will appear to you that using Fork/Join is a complex task. Once you get familiar with it, however, you’ll realize that it is conceptually easy and that it significantly simplifies your job. The key is to recursively subdivide the task into smaller chunks that can be processed by separate threads.

Visualizes how the task is recursively subdivided into smaller tasks and how the partial results are combined. As shown by the figure, a task is split into two subtasks, and then each subtask is again split in two subtasks, and so on until each split subtask is computable by each thread. Once a thread completes the computation, it returns the result for combining it with other results; in this way all the computed results are combined back.

How the Fork/Join framework uses divide-and-conquer to complete the task.

Briefly, the Fork/Join algorithm is designed as follows:

forkJoinAlgorithm() {

split tasks;

fork the tasks;

join the tasks;

compose the results;

}

Here is the pseudo-code of how these steps work:

doRecursiveTask(input) {

if (the task is small enough to be handled by a thread) {

compute the small task;

if there is a result to return, do so

}

else {

divide (i.e., fork) the task into two parts

call compute() on first task, join() on second task, combine both results and return

}

}

Useful Classes of the Fork/Join Framework the following classes play key roles in the Fork/Join framework: ForkJoinPool, ForkJoinTask, RecursiveTask, and RecursiveAction. Let’s consider these classes in more detail.

To provide effective parallel execution, the fork/join framework uses a pool of threads called the ForkJoinPool, which manages worker threads of type ForkJoinWorkerThread.

The ForkJoinPool is the heart of the framework. It is an implementation of the ExecutorService that manages worker threads and provides us with tools to get information about the thread pool state and performance.

Worker threads can execute only one task at the time, but the ForkJoinPool doesn’t create a separate thread for every single subtask. Instead, each thread in the pool has its own double-ended queue which stores tasks. This architecture is vital for balancing the thread’s workload with the help of the work-stealing algorithm  which is simply free threads try to “steal” work from deques of busy threads

Important Methods in the ForkJoinPool Class.
void execute(ForkJoinTask<?> task) Executes a given task asynchronously.

<T> T invoke(ForkJoinTask<T> task) Executes the given task and returns the computed result.

<T> List<Future<T>>invokeAll(Collection<? extends Callable<T>> tasks) Executes all the given tasks and returns a list of future objects when all the tasks are completed.

boolean isTerminated() Returns true if all the tasks are completed.

int getParallelism()  Status checking methods.

int getPoolSize()

long getStealCount()

int getActiveThreadCount()

<T> ForkJoinTask<T> submit(Callable<T> task) Executes a submitted task. Overloaded versions take different types of tasks; returns a Task object or a Future object.

<T> ForkJoinTask<T> submit(ForkJoinTask<T> task)

ForkJoinTask<?> submit(Runnable task)

<T> ForkJoinTask<T> submit(Runnable task, T result)

In Java 8, the most convenient way to get access to the instance of the ForkJoinPool is to use its static method commonPool(). As its name suggests, this will provide a reference to the common pool, which is a default thread pool for every ForkJoinTask.

According to Oracle’s documentation, using the predefined common pool reduces resource consumption, since this discourages the creation of a separate thread pool per task.

ForkJoinPool commonPool = ForkJoinPool.commonPool();

In Java 7

Now it can be easily accessed

ForkJoinPool forkJoinPool = PoolUtil.forkJoinPool;

With ForkJoinPool’s constructors, it is possible to create a custom thread pool with a specific level of parallelism, thread factory, and exception handler. In the example above, the pool has a parallelism level of 2. This means that pool will use 2 processor cores.

ForkJoinTask<V> is a lightweight thread-like entity representing a task that defines methods

such as fork() and join(). ForkJoinTask is the base type for tasks executed inside ForkJoinPool.

In practice, one of its two subclasses should be extended: the RecursiveAction for void tasks and the RecursiveTask<V> for tasks that return a value.

They both have an abstract method compute() in which the task’s logic is defined.

Important Methods in the ForkJoinTask Class.

boolean cancel(boolean mayInterruptIfRunning) Attempts to cancel the execution of the task.

ForkJoinTask<V> fork() Executes the task asynchronously.

• V join() Returns the result of the computation when the computation is done.
• V get() Returns the result of the computation; waits if the computation is not complete.
• V invoke() Starts the execution of the submitted tasks; waits until computation complete, and returns results.

static <T extends ForkJoinTask<?>> Collection<T> invokeAll(Collection<T> tasks)

boolean isCancelled() Returns true if the task is cancelled.

boolean isDone() Returns true if the task is completed.

Let’s ascertain how you can use Fork/Join framework in problem solving. Here are the steps to use the framework:

First, check whether the problem is suitable for the Fork/Join framework or not.

Remember:

The Fork/Join framework is not suitable for all kinds of tasks. This framework is suitable if your problem fits this description:

The problem can be designed as a recursive task where the task can be subdivided into smaller units and the results can be combined together.

The subdivided tasks are independent and can be computed separately without the need for communication between the tasks when computation is in process. (Of course, after the computation is over, you will need to join them together.)

If the problem you want to solve can be modeled recursively, then define a task class that extends either RecursiveTask or RecursiveAction. If a task returns a result, extend from RecursiveTask; otherwise extend from RecursiveAction.Override the compute() method in the newly defined task class. The compute() method actually performs the task if the task is small enough to be executed; or split the task into subtasks and invoke them. The subtasks can be invoked either by invokeAll() or fork() method (use fork() when the subtask returns a value). Use the join() method to get the computed results (if you used fork() method earlier).

Merge the results, if computed from the subtasks. Then instantiate ForkJoinPool, create an instance of the task class, and start the execution of the task using the invoke() method on the ForkJoinPool instance.

Now let’s try solving the problem of how to sum 1..N where N is a large number. We subdivided the sum computation task iteratively into ten sub-ranges; then you computed the sum for each sub-range and then computed the sum-of-the-partial sums. Alternatively, you can solve this problem rescursively using the Fork/Join framework.

Example have attached.

Now we used RecursiveTask; however, if a task is not returning a value, then we should use RecursiveAction. Let’s implement a search program using RecursiveAction. Assume that you have a big array (say of 10,000 items) and we want to search a key item. You can use the Fork/Join framework to split the task into several subtasks and execute them in parallel.

Example have attached.

generate class diagram from java code eclipse

Generate class diagram from java code online

Below have script to generate online flow diagram of application

to generate online diagram click on given link and paste the script click

title saveAgreementConsent

participant Client

participant Enterprise Service

participant Agreement Service

participant OSB

Client -> Enterprise Service: saveAgreementConsent Request

note right of Client

Method Type: POST

{

  "accountNumber": "string",

  "agreementId": 0,

  "agreementName": "string",

  "agreementType": "string",

  "confirm": true,

  "transactionId": "string"

  "divisionId": "string"

}

end note

Enterprise Service -> Enterprise Service: check "agreementId"

alt agreementId == 1 i.e., agreementType == E911

note over Client, Enterprise Service, Agreement Service: 'Flow A' starts

Enterprise Service -> Agreement Service: Invoke updateAgreement operation

Agreement Service --> Enterprise Service: updateAgreement Response

Enterprise Service --> Client: saveAgreementsConsent Response

note over Client, Enterprise Service, Agreement Service: 'Flow A' stops

end

alt agreementId == 2 i.e., agreementType == T&C

Enterprise Service -> Enterprise Service: Repeat 'Flow A'

end

alt agreementId == 3 i.e., agreementType == PrivacyPolicy

Enterprise Service -> SPC Service: getDivisionID

SPC Service--> Enterprise Service: Response (MSO)

alt if MSO=="L-CHTR"

Enterprise Service -> OSB: Invoke createAccountAgreementsConsent operation (OSB)

OSB --> Enterprise Service: createAccountAgreementsConsent Response

Enterprise Service --> Client: saveAgreementConsent Response

end

alt if MSO=="L-BHN" || "L-TWC"

Enterprise Service -> Enterprise Service: Repeat 'Flow A'

Enterprise Service --> Client: saveAgreementConsent Response

end

end

Callable Executors ExecutorService ThreadPool and Future class

Callable, Executors, ExecutorService, ThreadPool, and Future

Callable is an interface that declares only one method: call(). Its full signature is V call() throws Exception. It represents a task that needs to be completed by a thread. Once the task completes, it returns a value. For some reason, if the call() method cannot execute or fails, it throws an Exception.

To execute a task using the Callable object, you first create a thread pool. A thread pool is a collection of threads that can execute tasks. You create a thread pool using the Executors utility class. This class provides methods to get instances of thread pools, thread factories, etc.

The ExecutorService interface implements the Executor interface and provides services such as termination of threads and production of Future objects. Some tasks may take considerable execution time to complete. So, when you submit a task to the executor service, you get a Future object.

Future represents objects that contain a value that is returned by a thread in the future (i.e., it returns the value once the thread terminates in the “future”). You can use the isDone() method in the Future class to check if the task is complete and then use the get() method to fetch the task result. If you call the get() method directly while the task is not complete, the method blocks until it completes and returns the value once available.

import java.util.concurrent.Callable;

public class Factorial implements Callable<Long> {

long n;

public Factorial(long n) { 

this.n = n;        

}

public Long call() throws Exception {

if (n <= 0) {

throw new Exception("for finding factorial, N should be > 0");

}

long fact = 1;

for (long longVal = 1; longVal <= n; longVal++) {

fact *= longVal;

}

return fact;

}

}

import java.util.concurrent.Callable;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.Future;

public class CallableTest {

public static void main(String []args) throws Exception {             

// the value for which we want to find the factorial                

long N = 4;                

// get a callable task to be submitted to the executor service                

Callable<Long> task = new Factorial(N);                

// create an ExecutorService with a fixed thread pool consisting of one thread                

ExecutorService es = Executors.newSingleThreadExecutor();                

// submit the task to the executor service and store the Future object                

Future<Long> future = es.submit(task);                

// wait for the get() method that blocks until the computation is complete.                

System.out.printf("factorial of %d is %d", N, future.get());                

// done. shutdown the executor service since we don't need it anymore                

if(!es.isShutdown())

es.shutdown();  

System.out.println(" and Sevice is shutdown");

} 

}

In this program, you have a Factorial class that implements Callable. Since the task is to compute the factorial of a number N, the task needs to return a result. You use Long type for the factorial value, so you implement Callable<Long>. Inside the Factorial class, you define the call() method that actually performs the task (the task

here is to compute the factorial of the given number). If the given value N is negative or zero, you don’t perform the task and throw an exception to the caller. Otherwise, you loop from 1 to N and find the factorial value.

In the CallableTest class, you first create an instance of the Factorial class. You then need to execute this task. 

For the sake of simplicity, you get a singled-threaded executor by calling the newSingleThreadExecutor() method in the Executors class. Note that you could use other methods such as newFixedThreadPool(nThreads) to create a thread pool with multiple threads depending on the level of parallelism you need.Once you get an ExecutorService, you submit the task for execution. ExecutorService abstracts details such as when the task is executed, how the task is assigned to the threads, etc. You get a reference to Future<Long> when you call the submit(task) method. From this future reference, you call the get() method to fetch the result after completing the task. If the task is still executing when you call future.get(), this get() method will block until the task execution completes. Once the execution is complete, you need to manually release the ExecutorService by calling the shutdown() method.Now that you are familiar with the basic mechanism of how to execute tasks, here’s a complex example. Assume that your task is to find the sum of numbers from 1 to N where N is a large number (a million in our case). Of course, you can use the formula [(N * (N + 1)) / 2] to find out the sum. Yes, you’ll make use of this formula to check if the summation from 1 . . . N is correct or not. However, just for illustration, you’ll divide the range 1 to 1 million to N sub-ranges and by spawn N threads to sum up numbers in that sub-range.

import java.util.*;

import java.util.concurrent.*;

//We create a class SumOfN that sums the values from 1..N where N is a large number.

//We divide the task // to sum the numbers to 10 threads (which is an arbitrary limit just for illustration).

//Once computation is complete, we add the results of all the threads,

//and check if the calculation is correct by using the formula (N * (N + 1))/2.

class SumOfN {

            private static long N = 1_000_000L;

            // one million

            private static long calculatedSum = 0;

           

            // value to hold the sum of values in range 1..N

            private static final int NUM_THREADS = 10;

            // number of threads to create for distributing the effort

            // This Callable object sums numbers in range from..to

static class SumCalc implements Callable<Long> {               

                                     long from, to, localSum = 0;

                                     

             public SumCalc(long from, long to) {                       

                                                this.from = from;                       

                                                this.to = to;               

                                                }

            

                                    public Long call() {                       

                                                // add in range 'from' .. 'to' inclusive of the value 'to'                       

                                                for(long i = from; i <= to; i++) {                               

                                                localSum += i;                       

                                                }                       

                                                return localSum;                

                                                }       

            }

//In the main method we implement the logic to divide the summation tasks to

//given number of threads and finally check if the calculated sum is correct

public static void main(String []args) {

//Divide the task among available fixed number of threads

ExecutorService executorService = Executors.newFixedThreadPool(NUM_THREADS);

//store the references to the Future objects in a List for summing up together

List<Future<Long>> summationTasks = new ArrayList<>();

long nByTen = N/10; // divide N by 10 so that it can be submitted as 10 tasks

for(int i = 0; i < NUM_THREADS; i++) {

//create a summation task

//starting from (10 * 0) + 1 .. (N/10 * 1) to (10 * 9) + 1 .. (N/10 * 10)

long fromInInnerRange = (nByTen * i) + 1;

long toInInnerRange = nByTen * (i+1);

System.out.printf("Spawning thread for summing in range %d to %d %n",

fromInInnerRange, toInInnerRange);

//Create a callable object for the given summation range

Callable<Long> summationTask =

new SumCalc(fromInInnerRange, toInInnerRange);

//submit that task to the executor service

Future<Long> futureSum = executorService.submit(summationTask);

//it will take time to complete, so add it to the list to revisit later

summationTasks.add(futureSum);

}

executorService.shutdown();

//now, find the sum from each task

for(Future<Long> partialSum : summationTasks) {

try {

//the get() method will block (i.e., wait) until the computation is over

calculatedSum += partialSum.get();

} catch(CancellationException | ExecutionException

| InterruptedException exception) {

//unlikely that you get an exception - exit in case something goes wrong

exception.printStackTrace();

System.exit(-1);

}

}

//now calculate the sum using formula (N * (N + 1))/2 without doing the hard-work

long formulaSum = (N * (N + 1))/2;

//print the sum using formula and the ones calculated one by one

//they must be equal!

System.out.printf("Sum by threads = %d, sum using formula = %d",

calculatedSum, formulaSum);

}

}

/*

Spawning thread for summing in range 1 to 100000000

Spawning thread for summing in range 100000001 to 200000000

Spawning thread for summing in range 200000001 to 300000000

Spawning thread for summing in range 300000001 to 400000000

Spawning thread for summing in range 400000001 to 500000000

Spawning thread for summing in range 500000001 to 600000000

Spawning thread for summing in range 600000001 to 700000000

Spawning thread for summing in range 700000001 to 800000000

Spawning thread for summing in range 800000001 to 900000000

Spawning thread for summing in range 900000001 to 1000000000

Sum by threads = 500000000500000000, sum using formula = 500000000500000000

Let’s now analyze how this program works. In this program, you need to find the sum of 1..N where N is one

million (a large number). The class SumCalc implements Callable<Long> to sum the values in the range from to

to. The call() method performs the actual computation of the sum by looping from from to to and returns the

intermediate sum value as a Long value.

In this program, you divide the summation task among multiple threads. You can determine the number of

threads based on the number of cores available in your processor; however, for the sake of keeping the program

simpler, use ten threads.

In the main() method, you create a ThreadPool with ten threads. You are going to create ten summation tasks, so

you need a container to hold the references to those tasks. Use ArrayList to hold the Future<Long> references.

In the first for loop in main(), you create ten tasks and submit them to the ExecutorService. As you submit a

task, you get a Future<Long> reference and you add it to the ArrayList.

Once you’ve created the ten tasks, you traverse the array list in the next for loop to get the results of the tasks. You

sum up the partial results of the individual tasks to compute the final sum.

Once you get the computed sum of values from one to one million, you use the simple formula N * (N + 1)/2

to find the formula sum. From the output, you can see that the computed sum and the formula sum are equal, so you

can ascertain that your logic of dividing the tasks and combining the results of the tasks worked correctly.*/

ThreadFactory

ThreadFactory is an interface that is meant for creating threads instead of explicitly creating threads by calling new Thread(). For example, assume that you often create high-priority threads. You can create a MaxPriorityThreadFactory to set the default priority of threads created by that factory to maximum priority

This will use when we wanna set some thread priority high so we can use below example for standard way.

import java.util.concurrent.ThreadFactory;

public class MaxPriorityThreadFactory implements ThreadFactory {

            private static long count = 0;

            public Thread newThread(Runnable r) {

            Thread temp = new Thread(r);

            temp.setName("prioritythread" + count++);

            temp.setPriority(Thread.MAX_PRIORITY);

            return temp;

            }

            }

public class ARunnable implements Runnable {

            public void run() {

                        System.out.println("Running the created thread ");

                        }

                        }

public class TestThreadFactory {

            public static void main(String []args) {

                        ThreadFactory threadFactory = new MaxPriorityThreadFactory();

                        ThreadFactory threadFactory1 = new MaxPriorityThreadFactory();

                        Thread t1 = threadFactory.newThread(new ARunnable());

                        System.out.println("The name of the thread is " + t1.getName());

                        System.out.println("The priority of the thread is " + t1.getPriority());

                        t1.start();

                        }

                        }

The ThreadLocalRandom Class When you do concurrent programming, you’ll find that there is often a need to generate random numbers.Using Math.random() is not efficient for concurrent programming. For this reason, the java.util.concurrent package introduces the ThreadLocalRandom class, which is suitable for use in concurrent programs. You can use ThreadLocalRandom.current() and then call methods such as nextInt() and nextFloat() to generate the random numbers.