In this example, we shall be demonstrating the use of java.util.concurrent.ForkJoinPool Class. This class was introduced in Java 7.
The java.util.concurrent.ForkJoinPool Class implements java.util.concurrent.Executor and java.util.concurrent.ExecutorService interfaces. This class works on divide-and-conquer policy. Each major subtask is divided into a number of sub-tasks and a new thread is spawned for each new sub-task. Recursion is the most popular strategy for dividing the work. Once sub-tasks are completed(conquered), their results are joined together and the final result is returned.
Unlike other implementations of
ExecutorService, java.util.concurrent.ForkJoinPool need not be explicitly shut-down as all threads in this pool are started as Daemon Threads.The sub-tasks created are usually sub-classes of either RecursiveAction or RecursiveTask abstract classes. The task class have to override the compute method. Upon invocation of the task the compute method is invoked. The difference between the two classes is that the compute method of RecursiveTask<V> Class returns the result of computation from the sub-tasks. (analogous to the Callable)
1. Pseudo-code for Fork/Join high-level view
Result solve(Problem problem)
{
if (problem is smaller than threshhold)
solve the problem sequentially
else {
split the task into sub-tasks
fork new subtasks to solve each part
join all subtasks to get the sub-results
combine the sub-results to get the final result.
}
}
We will implement the RecursiveTask<V> Class and create the tasks for searching a given folder and its sub-folders, for a given string in the name of the files. The tasks are implemented in its compute() method. The main task will spawn other tasks recursively, i.e. one task for each sub-folder in the given main folder. The After all the sub-tasks are forked, we shall join them to get the final List of files(file names). The ForkJoinPoolDemo class then creates a ForkJoinPool object, which will start executing the main task.
In case, we do not specify any argument while creating the object of ForkJoinPool the size of pool is equivalent to Runtime.getRuntime().availableProcessors() i.e. the available number of processors(counting the ability of hyper threading by processors) on the machine.
SearchDirectory.java:
package com.javacodegeeks.examples.concurrent;
import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.RecursiveTask;
/**
* @author Chandan Singh
*/
public class SearchDirectory extends RecursiveTask<List<String>>
{
private static final long serialVersionUID = 7570286838840187368L;
private String directoryName = "";
private String searchString = "";
public SearchDirectory(String directoryName, String searchString)
{
this.directoryName = directoryName;
this.searchString = searchString;
}
@Override
protected List<String> compute()
{
List<String> matchingFilesList = new ArrayList<>();
List<SearchDirectory> taskList = new ArrayList<>();
File directory = new File(directoryName);
if(directoryName == null || "".equals(directoryName) || !directory.exists())
throw new IllegalArgumentException("Directory Name is NOT Valid");
File[] fileArray = directory.listFiles();
for(File file : fileArray)
{
if(file.isDirectory())
{
SearchDirectory searchDirectory = new SearchDirectory(directoryName,searchString);
searchDirectory.fork();
taskList.add(searchDirectory);
}
else
{
if(checkName(file.getName()))
matchingFilesList.add(file.getPath());
}
}
for(SearchDirectory sd : taskList)
{
List<String> intermediateResultList = sd.join();
matchingFilesList.addAll(intermediateResultList);
}
return matchingFilesList;
}
private boolean checkName(String filename)
{
return filename.contains(searchString);
}
}
ForkJoinPoolExample.java:
package com.javacodegeeks.examples.concurrent;
import java.util.List;
import java.util.concurrent.ForkJoinPool;
/**
* @author Chandan Singh
*/
public class ForkJoinPoolExample
{
public static void main(String[] args)
{
ForkJoinPool pool = new ForkJoinPool();
SearchDirectory searchDir = new SearchDirectory("F:\\vapt\\SFMSNEFT","NEFT");
pool.execute(searchDir);
List<String> fileList = searchDir.join();
System.out.println("The Search returned following files : "+fileList);
}
}
2. Application of ForkJoinPool in Existing Core Java Packages
ForkJoinPoolis used injava.util.Arrays#parallelSortmethods.java.util.Stream#parallel()also uses theForkJoinPool
3. Conclusion
Thus we have studied the working of ForkJoinPool in Java and its benefits over other ExecutorService implementations.
You can download the source code of this example here: ForkJoinPoolExample.zip
