ThreadAdavncedPoolGroupForkJoin.txt

==================Java Thread Pool=================

Java Thread pool represents a group of worker threads that are waiting 
for the job and reuse many times.

In case of thread pool, a group of fixed size threads are created. 
A thread from the thread pool is pulled out and assigned a job by
the service provider. 
After completion of the job, thread is contained in the thread pool again.


Advantage of Java Thread Pool

Better performance It saves time because there is no need to create 
new thread.
Real time usage

It is used in Servlet and JSP where container creates a thread pool to process the request.
Example of Java Thread Pool

Let's see a simple example of java thread pool using ExecutorService and Executors.

File: WorkerThread.java

    import java.util.concurrent.ExecutorService;  
    import java.util.concurrent.Executors;  
    class WorkerThread implements Runnable {  
        private String message;  
        public WorkerThread(String s){  
            this.message=s;  
        }  
         public void run() {  
            System.out.println(Thread.currentThread().getName()+" (Start) message = "+message);  
            processmessage();//call processmessage method that sleeps the thread for 2 seconds  
            System.out.println(Thread.currentThread().getName()+" (End)");//prints thread name  
        }  
        private void processmessage() {  
            try {  Thread.sleep(2000);  } catch (InterruptedException e) { e.printStackTrace(); }  
        }  
    }  

File: JavaThreadPoolExample.java

    public class TestThreadPool {  
         public static void main(String[] args) {  
            ExecutorService executor = Executors.newFixedThreadPool(5);//creating a pool of 5 threads  
            for (int i = 0; i < 10; i++) {  
                Runnable worker = new WorkerThread("" + i);  
                executor.execute(worker);//calling execute method of ExecutorService  
              }  
            executor.shutdown();  
            while (!executor.isTerminated()) {   }  
      
            System.out.println("Finished all threads");  
        }  
     }  


Output:

pool-1-thread-1 (Start) message = 0
pool-1-thread-2 (Start) message = 1
pool-1-thread-3 (Start) message = 2
pool-1-thread-5 (Start) message = 4
pool-1-thread-4 (Start) message = 3
pool-1-thread-2 (End)
pool-1-thread-2 (Start) message = 5
pool-1-thread-1 (End)
pool-1-thread-1 (Start) message = 6
pool-1-thread-3 (End)
pool-1-thread-3 (Start) message = 7
pool-1-thread-4 (End)
pool-1-thread-4 (Start) message = 8
pool-1-thread-5 (End)
pool-1-thread-5 (Start) message = 9
pool-1-thread-2 (End)
pool-1-thread-1 (End)
pool-1-thread-4 (End)
pool-1-thread-3 (End)
pool-1-thread-5 (End)
Finished all threads

===============ThreadGroup in Java============================


Java provides a convenient way to group multiple threads in a 
single object. In such way, we can suspend, resume or interrupt group 
of threads by a single method call.
Note: Now suspend(), resume() and stop() methods are deprecated.

Java thread group is implemented by java.lang.ThreadGroup class.

A ThreadGroup represents a set of threads. 
A thread group can also include the other thread group. 
The thread group creates a tree in which every thread group except
 the initial thread group has a parent.

A thread is allowed to access information about its own thread group, but it cannot access the information about its thread group's parent thread group or any other thread groups.
Constructors of ThreadGroup class

There are only two constructors of ThreadGroup class.
No.	Constructor	Description
1)	ThreadGroup(String name)	creates a thread group with given name.
2)	ThreadGroup(ThreadGroup parent, String name)	creates a thread group with given parent group and name.
Methods of ThreadGroup class

There are many methods in ThreadGroup class. A list of ThreadGroup methods are given below.
S.N. 	Modifier and Type 	Method 	Description
1) 	void 	checkAccess() 	This method determines if the currently running thread has permission to modify the thread group.
2) 	int 	activeCount() 	This method returns an estimate of the number of active threads in the thread group and its subgroups.
3) 	int 	activeGroupCount() 	This method returns an estimate of the number of active groups in the thread group and its subgroups.
4) 	void 	destroy() 	This method destroys the thread group and all of its subgroups.
5) 	int 	enumerate(Thread[] list) 	This method copies into the specified array every active thread in the thread group and its subgroups.
6) 	int 	getMaxPriority() 	This method returns the maximum priority of the thread group.
7) 	String 	getName() 	This method returns the name of the thread group.
8) 	ThreadGroup 	getParent() 	This method returns the parent of the thread group.
9) 	void 	interrupt() 	This method interrupts all threads in the thread group.
10) 	boolean 	isDaemon() 	This method tests if the thread group is a daemon thread group.
11) 	void 	setDaemon(boolean daemon) 	This method changes the daemon status of the thread group.
12) 	boolean 	isDestroyed() 	This method tests if this thread group has been destroyed.
13) 	void 	list() 	This method prints information about the thread group to the standard output.
14) 	boolean 	parentOf(ThreadGroup g 	This method tests if the thread group is either the thread group argument or one of its ancestor thread groups.
15) 	void 	suspend() 	This method is used to suspend all threads in the thread group.
16) 	void 	resume() 	This method is used to resume all threads in the thread group which was suspended using suspend() method.
17) 	void 	setMaxPriority(int pri) 	This method sets the maximum priority of the group.
18) 	void 	stop() 	This method is used to stop all threads in the thread group.
19) 	String 	toString() 	This method returns a string representation of the Thread group.

Let's see a code to group multiple threads.

    ThreadGroup tg1 = new ThreadGroup("Group A");   
    Thread t1 = new Thread(tg1,new MyRunnable(),"one");     
    Thread t2 = new Thread(tg1,new MyRunnable(),"two");     
    Thread t3 = new Thread(tg1,new MyRunnable(),"three");    

Now all 3 threads belong to one group. Here, tg1 is the thread group name, MyRunnable is the class that implements Runnable interface and "one", "two" and "three" are the thread names.

Now we can interrupt all threads by a single line of code only.

    Thread.currentThread().getThreadGroup().interrupt();  

ThreadGroup Example

File: ThreadGroupDemo.java

    public class ThreadGroupDemo implements Runnable{  
        public void run() {  
              System.out.println(Thread.currentThread().getName());  
        }  
       public static void main(String[] args) {  
          ThreadGroupDemo runnable = new ThreadGroupDemo();  
              ThreadGroup tg1 = new ThreadGroup("Parent ThreadGroup");  
                
              Thread t1 = new Thread(tg1, runnable,"one");  
              t1.start();  
              Thread t2 = new Thread(tg1, runnable,"two");  
              t2.start();  
              Thread t3 = new Thread(tg1, runnable,"three");  
              t3.start();  
                   
              System.out.println("Thread Group Name: "+tg1.getName());  
             tg1.list();  
      
        }  
       }  

Output:

one
two
three
Thread Group Name: Parent ThreadGroup
java.lang.ThreadGroup[name=Parent ThreadGroup,maxpri=10]
    Thread[one,5,Parent ThreadGroup]
    Thread[two,5,Parent ThreadGroup]
    Thread[three,5,Parent ThreadGroup]

=======================Java Shutdown Hook====================

The shutdown hook can be used to perform cleanup resource or save the state when JVM shuts down normally or abruptly. Performing clean resource means closing log file, sending some alerts or something else. So if you want to execute some code before JVM shuts down, use shutdown hook.
When does the JVM shut down?
The JVM shuts down when:

    user presses ctrl+c on the command prompt
    System.exit(int) method is invoked
    user logoff
    user shutdown etc.

The addShutdownHook(Thread hook) method

The addShutdownHook() method of Runtime class is used to register the thread with the Virtual Machine. Syntax:

    public void addShutdownHook(Thread hook){}  


The object of Runtime class can be obtained by calling the static factory method getRuntime(). For example:

Runtime r = Runtime.getRuntime();

Factory method

The method that returns the instance of a class is known as factory method.
Simple example of Shutdown Hook

    class MyThread extends Thread{  
        public void run(){  
            System.out.println("shut down hook task completed..");  
        }  
    }  
      
    public class TestShutdown1{  
    public static void main(String[] args)throws Exception {  
      
    Runtime r=Runtime.getRuntime();  
    r.addShutdownHook(new MyThread());  
          
    System.out.println("Now main sleeping... press ctrl+c to exit");  
    try{Thread.sleep(3000);}catch (Exception e) {}  
    }  
    }  

Output:Now main sleeping... press ctrl+c to exit
       shut down hook task completed..
       
 

Note: The shutdown sequence can be stopped by invoking the halt(int) method of Runtime class.
Same example of Shutdown Hook by annonymous class:

    public class TestShutdown2{  
    public static void main(String[] args)throws Exception {  
      
    Runtime r=Runtime.getRuntime();  
      
    r.addShutdownHook(new Thread(){  
    public void run(){  
        System.out.println("shut down hook task completed..");  
        }  
    }  
    );  
          
    System.out.println("Now main sleeping... press ctrl+c to exit");  
    try{Thread.sleep(3000);}catch (Exception e) {}  
    }  
    }  

Output:Now main sleeping... press ctrl+c to exit
       shut down hook task completed..
       
 ==================How to perform single task by multiple threads?========
 
If you have to perform single task by many threads, have only one run() method.For example:
Program of performing single task by multiple threads

    class TestMultitasking1 extends Thread{  
     public void run(){  
       System.out.println("task one");  
     }  
     public static void main(String args[]){  
      TestMultitasking1 t1=new TestMultitasking1();  
      TestMultitasking1 t2=new TestMultitasking1();  
      TestMultitasking1 t3=new TestMultitasking1();  
      
      t1.start();  
      t2.start();  
      t3.start();  
     }  
    }  

Test it Now

Output:task one
       task one
       task one

Program of performing single task by multiple threads

    class TestMultitasking2 implements Runnable{  
    public void run(){  
    System.out.println("task one");  
    }  
      
    public static void main(String args[]){  
    Thread t1 =new Thread(new TestMultitasking2());//passing annonymous object of TestMultitasking2 class  
    Thread t2 =new Thread(new TestMultitasking2());  
      
    t1.start();  
    t2.start();  
      
     }  
    }  

Test it Now

Output:task one
       task one

Note: Each thread run in a separate callstack.
MultipleThreadsStack
How to perform multiple tasks by multiple threads (multitasking in multithreading)?
If you have to perform multiple tasks by multiple threads,have multiple run() methods.For example:
Program of performing two tasks by two threads

    class Simple1 extends Thread{  
     public void run(){  
       System.out.println("task one");  
     }  
    }  
      
    class Simple2 extends Thread{  
     public void run(){  
       System.out.println("task two");  
     }  
    }  
      
     class TestMultitasking3{  
     public static void main(String args[]){  
      Simple1 t1=new Simple1();  
      Simple2 t2=new Simple2();  
      
      t1.start();  
      t2.start();  
     }  
    }  

Test it Now

Output:task one
       task two

Same example as above by annonymous class that extends Thread class:
Program of performing two tasks by two threads

    class TestMultitasking4{  
     public static void main(String args[]){  
      Thread t1=new Thread(){  
        public void run(){  
          System.out.println("task one");  
        }  
      };  
      Thread t2=new Thread(){  
        public void run(){  
          System.out.println("task two");  
        }  
      };  
      
      
      t1.start();  
      t2.start();  
     }  
    }  

Test it Now

Output:task one
       task two

Same example as above by annonymous class that implements Runnable interface:
Program of performing two tasks by two threads

    class TestMultitasking5{  
     public static void main(String args[]){  
      Runnable r1=new Runnable(){  
        public void run(){  
          System.out.println("task one");  
        }  
      };  
      
      Runnable r2=new Runnable(){  
        public void run(){  
          System.out.println("task two");  
        }  
      };  
          
      Thread t1=new Thread(r1);  
      Thread t2=new Thread(r2);  
      
      t1.start();  
      t2.start();  
     }  
    }  

Test it Now

Output:task one
       task two
	   
	   ===========================ava Garbage Collection

In java, garbage means unreferenced objects.

Garbage Collection is process of reclaiming the runtime unused memory automatically. In other words, it is a way to destroy the unused objects.

To do so, we were using free() function in C language and delete() in C++. But, in java it is performed automatically. So, java provides better memory management.
Advantage of Garbage Collection

    It makes java memory efficient because garbage collector removes the unreferenced objects from heap memory.
    It is automatically done by the garbage collector(a part of JVM) so we don't need to make extra efforts.

How can an object be unreferenced?

There are many ways:

    By nulling the reference
    By assigning a reference to another
    By anonymous object etc.

Java Garbage Collection Scenario
1) By nulling a reference:

    Employee e=new Employee();  
    e=null;  

2) By assigning a reference to another:

    Employee e1=new Employee();  
    Employee e2=new Employee();  
    e1=e2;//now the first object referred by e1 is available for garbage collection  

3) By anonymous object:

    new Employee();  

finalize() method

The finalize() method is invoked each time before the object is garbage collected. This method can be used to perform cleanup processing. This method is defined in Object class as:

    protected void finalize(){}  

Note: The Garbage collector of JVM collects only those objects that are created by new keyword. So if you have created any object without new, you can use finalize method to perform cleanup processing (destroying remaining objects).
gc() method

The gc() method is used to invoke the garbage collector to perform cleanup processing. The gc() is found in System and Runtime classes.

    public static void gc(){}  

Note: Garbage collection is performed by a daemon thread called Garbage Collector(GC). This thread calls the finalize() method before object is garbage collected.
Simple Example of garbage collection in java

    public class TestGarbage1{  
     public void finalize(){System.out.println("object is garbage collected");}  
     public static void main(String args[]){  
      TestGarbage1 s1=new TestGarbage1();  
      TestGarbage1 s2=new TestGarbage1();  
      s1=null;  
      s2=null;  
      System.gc();  
     }  
    }  

Test it Now

       object is garbage collected
       object is garbage collected

==============Fork Join===================================
    	
Understanding Java Fork-Join Framework with Examples

This tutorial helps you understand and experiment with Fork/Join framework, which is used by several new features in Java 7 and Java 8. You will be able to write programs that run tasks in parallel utilized multicore processors which are very popular today (perhaps your computer’s CPU has at least 2 or 4 cores, doesn’t it?).

Notice that parallel execution is different than concurrent execution:

- In parallel execution, each thread is executed in a separate processing core. Therefore, tasks are really executed in true parallel fashion.

- In concurrent execution, the threads are executed on a same core. That means tasks are actually executed in interleave fashion, sharing processing time of a processing core.
Don’t worry if you think parallel programming is complex and difficult, as you will see the Fork/Join framework makes it easy for programmers.

 

1. What is Fork/Join Framework?
Fork/Join framework is a set of APIs that allow programmers to take advantage of parallel execution supported by multicore processors. It uses ‘divide-and-conquer’ strategy: divide a very large problem into smaller parts, which in turn, the small part can be divided further into smaller ones, recursively until a part can be solved directly. This is called ‘fork’.

Then all parts are executed in parallel on multiple processing cores. The results of each part are ‘joined’ together to produce the final result. Hence the name of the framework ‘Fork/Join’.

The following pseudo code illustrates how the divide and conquer strategies work with Fork/Join framework:

if (problemSize < threshold)

                solve problem directly

else {

                break problem into subproblems

                recursively solve each problem

                combine the results

}
Fork/Join framework is added to JDK since Java 7 and improved in Java 8. It is used by several new features in the Java programming language, including Streams API and sorting an array in parallel.

Fork/Join framework simplifies parallel programming because:

- It simplifies thread creation. Threads are created and managed automatically.

- It automatically makes use of multiple processors so programs can scale to make use of available processors.
With support for true parallel execution, Fork/Join framework can significantly reduce computation time and increase performance in solving very large problems such as image processing, video processing, big data processing, etc.

One interesting point about Fork/Join framework: it uses a work stealing algorithm to balance the load among threads: if a worker thread runs out of things to do, it can steal tasks from other threads that are still busy.

 

2. Understand Fork/Join Framework’s API
The Fork/Join framework API is implemented in the java.util.concurrent package. At its core are the following 4 classes:

    ForkJoinTask<V>: an abstract class that defines a task that runs within a ForkJoinPool.

    ForkJoinPool: a thread pool that manages the execution of ForkJoinTasks.

    RecursiveAction: a ForkJoinTask’s subclass for tasks that don’t return values.

    RecursiveTask<V>: a ForkJoinTask’s subclass for tasks that return values.

Basically, you implement code for solving problems in a subclass of either RecursiveAction or RecursiveTask. And then submit the task to be executed by a ForkJoinPool, which handles everything from threads management to utilization of multicore processor.

Let’s dive deeper into each of these classes before going through some code examples.

 
ForkJoinTask<V>
This is the abstract base class for tasks that run within a ForkJoinPool. The type parameter V specifies the result type of the task. A ForkJoinTask is a thread-like entity that represents lightweight abstraction of a task, rather than an actual thread of execution. This mechanism allows a large number o tasks to be managed by a small number of actual threads in a ForkJoinPool. Its key methods are:

    final ForkJoinTask<V> fork()

    final V join()

    final V invoke()

The fork() method submits the task to execute asynchronously. This method return this (ForkJoinTask) and the calling thread continues to run.

The join() method waits until the task is done and returns the result.

The invoke() method combines fork() and join() in a single call. It starts the task, waits for it to end and return the result.

In addition, the ForkJoinTask class provides a couple of static methods for invoking more than one task at a time:

    static void invokeAll(ForkJoinTask<?> task1, ForkJoinTask<?> task2): execute two tasks.

    static void invokeAll(ForkJoinTask<?>… taskList): execute a list of tasks.

 

RecursiveAction:
This is a recursive ForkJoinTaskthat doesn’t return a result. “Recursive” means that the task can be split into subtasks of itself by divide-and-conquer strategy (you’ll see how to divide in the code examples in the next section below).

You must override its abstract method compute() in which computational code is put.

protected abstract void compute();
 

RecursiveTask<V>:
Similar to RecursiveAction, but a RecursiveTask returns a result whose type is specified by the type parameter V. You also must to put computational code by overriding the compute() method:

protected abstract V compute();
 

ForkJoinPool:
This class is the heart of Fork/Join framework. It’s responsible for the management of threads and execution of ForkJoinTasks. You must first have an instance of ForkJoinPool in order to execute ForkJoinTasks.

There are two ways for acquiring a ForkJoinPool instance. The first way creates a ForkJoinPool object using one of its constructors:

    ForkJoinPool(): creates a default pool that supports a level of parallelism equal to the number of processors available in the system.

    ForkJoinPool(int parallelism): creates a pool with a custom level of parallelism which must be greater than 0 and not more than the actual number of processors available.

The level of parallelism determines the number of threads that can execute concurrently. In other words, it determines the number of tasks that can be executed simultaneously - but cannot exceed the number of processors.

However, that doesn’t limit the number of tasks that can be managed by the pool. A ForkJoinPool can manage many more tasks than its level of parallelism.

The second way to acquire a ForkJoinPool instance is obtaining the common pool instance using the following ForkJoinPool’s static method:

public static ForkJoinPool commonPool()
The common pool is statically constructed and automatically available for use.

 

Execute ForkJoinTasks in a ForkJoinPool
After you have created an instance of ForkJoinPool, you can start executing a task using one of the following methods:

    <T> T invoke(ForkJoinTask<T> task): executes the specified task and returns its result upon completion. This call is synchronous, meaning that the calling thread waits until this method returns. For a resultless task (RecursiveAction), the type parameter Tis Void.

    void execute(ForkJoinTask<?> task): executes the specified task asynchronously - the calling code doesn’t wait for the task’s completion - it continues to run.

Alternatively, you can execute a ForkJoinTask by calling its own methods fork() or invoke(). In this case, the common pool will be used automatically, if the task is not already running within a ForkJoinPool.

A noteworthy point: ForkJoinPool uses daemon threads that are terminated when all user threads are terminated. That means you don’t have to explicitly shutdown a ForkJoinPool (though it is possible). In the case of common pool, calling shutdown() has no effect because the pool is always available for use.

Okay, that’s enough for the theory.

In the next email, you will see some examples in action.

Okay, that’s enough for the theory. It’s time to see some examples in action.

 

3. Example #1 - Using RecursiveAction
In this first example, you will learn how to use the Fork/Join framework to execute a task that doesn’t return a result, by extending the RecursiveAction class.

Suppose that we need to do a transformation on a very large array of numbers. For the sake of simplicity, the transformation is simply multiply every element in the array by a specified number. The following code is for the transformation task:

	
import java.util.concurrent.*;
 
/**
 * This class illustrates how to create a ForkJoinTask that does not return
 * a result.
 * @author www.codejava.net
 */
public class ArrayTransform extends RecursiveAction {
    int[] array;
    int number;
    int threshold = 100_000;
    int start;
    int end;
 
    public ArrayTransform(int[] array, int number, int start, int end) {
        this.array = array;
        this.number = number;
        this.start = start;
        this.end = end;
    }
 
    protected void compute() {
        if (end - start < threshold) {
            computeDirectly();
        } else {
            int middle = (end + start) / 2;
 
            ArrayTransform subTask1 = new ArrayTransform(array, number, start, middle);
            ArrayTransform subTask2 = new ArrayTransform(array, number, middle, end);
 
            invokeAll(subTask1, subTask2);
        }
    }
 
    protected void computeDirectly() {
        for (int i = start; i < end; i++) {
            array[i] = array[i] * number;
        }
    }
}
As you can see, this is a subclass of RecursiveAction and it implements the computation in the compute() method.

The array and number are passed from its constructor. The parameters start and end specify the range of elements in the array to be processed. This helps splitting the array into sub arrays if its size is greater than a threshold, otherwise perform the computation on the whole array directly.

Look at the code snippet in the else block in the compute() method:

	
protected void compute() {
    if (end - start < threshold) {
        computeDirectly();
    } else {
        int middle = (end + start) / 2;
 
        ArrayTransform subTask1 = new ArrayTransform(array, number, start, middle);
        ArrayTransform subTask2 = new ArrayTransform(array, number, middle, end);
 
        invokeAll(subTask1, subTask2);
    }
}
Here we divide the array into 2 parts and create two subtasks that process each. In turn, the subtask may be also divided further into smaller subtasks recursively until the size is less than the threshold, which invokes the computeDirectly() method.

And then you can execute the main task on a ForkJoinPool like this:


ArrayTransform mainTask = new ArrayTransform(array, number, 0, SIZE);
ForkJoinPool pool = new ForkJoinPool();
pool.invoke(mainTask);
or execute the task on the common pool:

	
ArrayTransform mainTask = new ArrayTransform(array, number, 0, SIZE);
mainTask.invoke();
Here’s the full source code of the test program:


import java.util.*;
import java.util.concurrent.*;
 
/**
 * This program demonstrates how to execute a resultless ForkJoinTask in
 * a ForkJoinPool
 * @author www.codejava.net
 */
public class ForkJoinRecursiveActionTest {
    static final int SIZE = 10_000_000;
    static int[] array = randomArray();
 
    public static void main(String[] args) {
 
        int number = 9;
 
        System.out.println("First 10 elements of the array before: ");
        print();
 
        ArrayTransform mainTask = new ArrayTransform(array, number, 0, SIZE);
        ForkJoinPool pool = new ForkJoinPool();
        pool.invoke(mainTask);
 
        System.out.println("First 10 elements of the array after: ");
        print();
    }
 
    static int[] randomArray() {
        int[] array = new int[SIZE];
        Random random = new Random();
 
        for (int i = 0; i < SIZE; i++) {
            array[i] = random.nextInt(100);
        }
 
        return array;
    }
 
    static void print() {
        for (int i = 0; i < 10; i++) {
            System.out.print(array[i] + ", ");
        }
        System.out.println();
    }
}
As you can see, we test with an array of 10 million elements that are randomly generated. As the array is too large, we print only the first 10 elements before and after the computation to see the effect:

	
First 10 elements of the array before:
42, 98, 43, 14, 9, 92, 33, 18, 18, 76,
First 10 elements of the array after:
378, 882, 387, 126, 81, 828, 297, 162, 162, 684,
 

This book helps you improve your Java programming skills to a new level: Effective Java (2nd Edition)  

 

4. Example #2 - Using RecursiveTask
In this second example, you will learn how to implement a task that returns a result. The following task counts the occurrences of even numbers in a large array:


	
import java.util.concurrent.*;
 
/**
 * This class illustrates how to create a ForkJoinTask that returns a result.
 * @author www.codejava.net
 */
public class ArrayCounter extends RecursiveTask<Integer> {
    int[] array;
    int threshold = 100_000;
    int start;
    int end;
 
    public ArrayCounter(int[] array, int start, int end) {
        this.array = array;
        this.start = start;
        this.end = end;
    }
 
    protected Integer compute() {
        if (end - start < threshold) {
            return computeDirectly();
        } else {
            int middle = (end + start) / 2;
 
            ArrayCounter subTask1 = new ArrayCounter(array, start, middle);
            ArrayCounter subTask2 = new ArrayCounter(array, middle, end);
 
            invokeAll(subTask1, subTask2);
 
 
            return subTask1.join() + subTask2.join();
        }
    }
 
    protected Integer computeDirectly() {
        Integer count = 0;
 
        for (int i = start; i < end; i++) {
            if (array[i] % 2 == 0) {
                count++;
            }
        }
 
        return count;
    }
}
As you can see, this class extends the RecursiveTask and overrides the compute() method that returns a result (an Integer in this case).

And note that we use the join() method to combine the results of subtasks:

1
	
return subTask1.join() + subTask2.join();
The test program is similar to the RecursiveAction example:

	
import java.util.*;
import java.util.concurrent.*;
 
/**
 * This program demonstrates how to execute a ForkJoinTask that returns
 * a result in a ForkJoinPool
 * @author www.codejava.net
 */
public class ForkJoinRecursiveTaskTest {
    static final int SIZE = 10_000_000;
    static int[] array = randomArray();
 
    public static void main(String[] args) {
 
        ArrayCounter mainTask = new ArrayCounter(array, 0, SIZE);
        ForkJoinPool pool = new ForkJoinPool();
        Integer evenNumberCount = pool.invoke(mainTask);
 
        System.out.println("Number of even numbers: " + evenNumberCount);
    }
 
    static int[] randomArray() {
        int[] array = new int[SIZE];
        Random random = new Random();
 
        for (int i = 0; i < SIZE; i++) {
            array[i] = random.nextInt(100);
        }
 
        return array;
    }
 
}
Run this program and you will see the output something like this:

1
	
Number of even numbers: 5000045
 

5. Example #3 - Experiment with Parallelism
In this last example, you will learn how the level of parallelism affects the computation time.

The ArrayCounter class is rewritten to have the threshold passed from constructor like this:

	
import java.util.concurrent.*;
 
/**
 * This class illustrates how to create a ForkJoinTask that returns a result.
 * @author www.codejava.net
 */
public class ArrayCounter extends RecursiveTask<Integer> {
    int[] array;
    int threshold;
    int start;
    int end;
 
    public ArrayCounter(int[] array, int start, int end, int threshold) {
        this.array = array;
        this.start = start;
        this.end = end;
        this.threshold = threshold;
    }
 
    protected Integer compute() {
        if (end - start < threshold) {
            return computeDirectly();
        } else {
            int middle = (end + start) / 2;
 
            ArrayCounter subTask1 = new ArrayCounter(array, start, middle, threshold);
            ArrayCounter subTask2 = new ArrayCounter(array, middle, end, threshold);
 
            invokeAll(subTask1, subTask2);
 
 
            return subTask1.join() + subTask2.join();
        }
    }
 
    protected Integer computeDirectly() {
        Integer count = 0;
 
        for (int i = start; i < end; i++) {
            if (array[i] % 2 == 0) {
                count++;
            }
        }
 
        return count;
    }
}
And in the test program, the level of parallelism and threshold are passed as arguments to the program:
	
import java.util.*;
import java.util.concurrent.*;
 
/**
 * This program allows you to easily test performance for ForkJoinPool
 * with different values of parallelism and threshold.
 * @author www.codejava.net
 */
public class ParallelismTest {
    static final int SIZE = 10_000_000;
 
    static int[] array = randomArray();
 
    public static void main(String[] args) {
        int threshold = Integer.parseInt(args[0]);
        int parallelism = Integer.parseInt(args[1]);
 
        long startTime = System.currentTimeMillis();
 
        ArrayCounter mainTask = new ArrayCounter(array, 0, SIZE, threshold);
        ForkJoinPool pool = new ForkJoinPool(parallelism);
        Integer evenNumberCount = pool.invoke(mainTask);
 
        long endTime = System.currentTimeMillis();
 
        System.out.println("Number of even numbers: " + evenNumberCount);
 
        long time = (endTime - startTime);
        System.out.println("Execution time: " + time + " ms");
    }
 
    static int[] randomArray() {
        int[] array = new int[SIZE];
        Random random = new Random();
 
        for (int i = 0; i < SIZE; i++) {
            array[i] = random.nextInt(100);
        }
 
        return array;
    }
 
}
This program allows you to easily test the performance with different values of parallelism and threshold. Note that it prints the execution time at the end. Try to run this program several times with different arguments and observe the execution time. Here are the suggested commands:

1
2
3
4
5
6
7
	
java ParallelismTest 1 100000
java ParallelismTest 2 100000
java ParallelismTest 3 100000
java ParallelismTest 4 100000
java ParallelismTest 2 500000
java ParallelismTest 4 500000
…
 

6. Conclusion
So far I have walked you through a lesson about Fork/Join framework. Here are the key points to remember:

- Fork/Join framework is designed to simplify parallel programming for Java programmers.

- ForkJoinPool is the heart of Fork/Join framework. It allows many ForkJoinTasks to be executed by a small number of actual threads, with each thread running on a separate processing core.

- You can obtain an instance of ForkJoinPool by either using its constructor or static method commonPool() that returns the common pool.

- ForkJoinTask is an abstract class that represents a task that is lighter weight than a normal thread. You implement the computation logic by overriding its compute() method.

- RecursiveAction is a ForkJoinTask that doesn’t return a result.

- RecursiveTask is a ForkJoinTask that returns a result.

- ForkJoinPool is different than other pools as it uses work stealing algorithm which allows a thread that runs out of things to do, to steal tasks from other threads that are still busy.

- Threads in ForkJoinPool are daemon. You don’t have to explicitly shutdown the pool.

- You can execute a ForkJoinTask either by invoking its own methods invoke() or fork(), or by submitting the task to a ForkJoinPool and then call invoke() or execute() on the pool.

- Calling invoke() or fork() on a ForkJoinTask will cause the task to run in the common pool, if it is not already running in a ForkJoinPool.

- Use the join() method on ForkJoinTasks to combine the results.

- The invoke() method waits for the task’s completion, but the execute() method does not.