[[199061]]

1. What is Fork/Join

The official definition given by Oracle is: Fork/Join framework is a multi-threaded processor that implements the ExecutorService interface. It can divide a large task into several small tasks for concurrent execution, making full use of available resources, thereby improving the execution efficiency of the application.

Fork/Join implements ExecutorService, so its tasks also need to be executed in the thread pool. The difference is that it uses a work stealing algorithm, where idle threads can steal tasks from fully loaded threads to help execute. (My personal understanding of work stealing is: since each thread in the thread pool has a queue, and the threads do not affect each other. So each thread gets a task from the head of its own task queue to execute. If a thread's task queue is empty at some point, and there are still tasks in the task queues of other threads, then this thread will take a task from the task queues of other threads to help execute. It's like stealing other people's work)

The core of the Fork/Join framework is the ForkJoinPool class that inherits AbstractExecutorService, which ensures the normal operation of the work-stealing algorithm and ForkJoinTask.

The following is the original text quoted from Oracle's official definition:

The fork/join framework is an implementation of the ExecutorService interface that helps you take advantage of multiple processors. It is designed for work that can be broken into smaller pieces recursively. The goal is to use all the available processing power to enhance the performance of your application.

As with any ExecutorService implementation, the fork/join framework distributes tasks to worker threads in a thread pool. The fork/join framework is distinct because it uses a work-stealing algorithm. Worker threads that run out of things to do can steal tasks from other threads that are still busy.

The center of the fork/join framework is the ForkJoinPool class, an extension of the AbstractExecutorService class. ForkJoinPool implements the core work-stealing algorithm and can execute ForkJoinTask processes.

2. Basic usage of Fork/Join

(1) Fork/Join base class

As mentioned above, Fork/Join is about splitting a large task into several small tasks, so the first step is of course to split the tasks, roughly as follows:

 if (this task is small enough) {
 Perform the tasks to be done
 } else {
 Split the task into two smaller parts
 Execute two small parts and wait for the execution results
 }

To implement FrokJoinTask, we need a base class that inherits RecursiveTask or RecursiveAction, and put the above code into the coupute method of the base class according to our business situation. Both RecursiveTask and RecursiveAction inherit FrokJoinTask. The difference between them is that RecursiveTask has a return value while RecursiveAction does not. The following is a demo I made to select the elements that still have "a" in a string list:

 @Override  
 protected List<String> compute() {
 // When the difference between end and start is less than the threshold, start the actual screening  
 if (end - this .start < threshold) {
        List<String> temp = list.subList( this .start, end);
 return temp.parallelStream().filter(s -> s.contains( "a" )).collect(Collectors.toList());
 } else {
 // If the difference between end and start is greater than the threshold  
 // Break the big task into two small tasks.  
 int middle = ( this .start + end) / 2 ;
        ForkJoinTest left = new ForkJoinTest(list, this .start, middle, threshold);
        ForkJoinTest right = new ForkJoinTest(list, middle, end, threshold);
 // Execute two "small tasks" in parallel  
 left.fork();
 right.fork();
 // Combine the results of the two "small tasks"  
        List<String> join = left.join();
        join.addAll(right.join());
 return join;
 }
 }

(2) Execution class

After the base class is completed, you can start calling. When calling, we first need the Fork/Join thread pool ForkJoinPool, then submit a ForkJoinTask to the thread pool and get the result. The input parameter of the submit method of ForkJoinPool is a ForkJoinTask, and the return value is also a ForkJoinTask. It provides a get method to get the execution result.

The code is as follows:

 ForkJoinPool pool = new ForkJoinPool();
 // Submit a decomposable ForkJoinTask task  
 ForkJoinTask<List<String>> future = pool.submit(forkJoinService);
 System.out.println(future.get());
 // Close the thread pool  
 pool.shutdown();

In this way, we have completed the development of a simple Fork/Join.

Tip: In Java 8, the parallelSort() method of java.util.Arrays and the encapsulated methods in the java.util.streams package also use Fork/Join. (Careful readers may notice that I also use stream in Fork/Join, so this Fork/Join is actually redundant because stream has already implemented Fork/Join. However, this is just a demo display and it doesn’t matter if it has no practical use.)

Quoting the official text:

One such implementation, introduced in Java SE 8, is used by the java.util.Arrays class for its parallelSort() methods. These methods are similar to sort(), but leverage concurrency via the fork/join framework. Parallel sorting of large arrays is faster than sequential sorting when run on multiprocessor systems.

Another implementation of the fork/join framework is used by methods in the java.util.streams package, which is part of Project Lambda scheduled for the Java SE 8 release.

Attached is the complete code for future reference:

1. Define an abstract class (for expansion, which has no practical use in this example and can be left undefined):

 import java.util.concurrent.RecursiveTask; 
 
 /**
 * Description: ForkJoin interface
 * Designer: jack
 * Date: 2017/8/3
 * Version: 1.0.0
 */  
 public   abstract   class ForkJoinService<T> extends RecursiveTask<T>{
 @Override  
 protected   abstract T compute();
 }

2. Define the base class

 import java.util.List;
 import java.util.stream.Collectors; 
 
 /**
 * Description: ForkJoin base class
 * Designer: jack
 * Date: 2017/8/3
 * Version: 1.0.0
 */  
 public   class ForkJoinTest extends ForkJoinService<List<String>> { 
 
 private   static ForkJoinTest forkJoinTest;
 private   int threshold; //Threshold  
 private List<String> list; //List to be split   
 
 private ForkJoinTest(List<String> list, int threshold) {
 this .list = list;
 this .threshold = threshold;
 } 
 
 @Override  
 protected List<String> compute() {
 // When the difference between end and start is less than the threshold, start the actual screening  
 if (list.size() < threshold) {
 return list.parallelStream().filter(s -> s.contains( "a" )).collect(Collectors.toList());
 } else {
 // If the difference between end and start is greater than the threshold, split the large task into two small tasks.  
 int middle = list.size() / 2 ;
            List<String> leftList = list.subList( 0 , middle);
            List<String> rightList = list.subList(middle, list.size());
            ForkJoinTest left = new ForkJoinTest(leftList, threshold);
            ForkJoinTest right = new ForkJoinTest(rightList, threshold);
 // Execute two "small tasks" in parallel  
 left.fork();
 right.fork();
 // Combine the results of the two "small tasks"  
            List<String> join = left.join();
            join.addAll(right.join());
 return join;
 }
 } 
 
 /**
 * Get the ForkJoinTest instance
 * @param list List to be processed
 * @param threshold threshold
 * @return ForkJoinTest instance
 */  
 public   static ForkJoinService<List<String>> getInstance(List<String> list, int threshold) {
 if (forkJoinTest == null ) {
 synchronized (ForkJoinTest. class ) {
 if (forkJoinTest == null ) {
                    forkJoinTest = new ForkJoinTest(list, threshold);
 }
 }
 }
 return forkJoinTest;
 }
 }

3. Execution class

 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.ForkJoinTask; 
 
 /**
 * Description: Fork/Join execution class
 * Designer: jack
 * Date: 2017/8/3
 * Version: 1.0.0
 */  
 public   class Test { 
 
 public   static   void main(String args[]) throws ExecutionException, InterruptedException { 
 
        String[] strings = { "a" , "ah" , "b" , "ba" , "ab" , "ac" , "sd" , "fd" , "ar" , "te" , "se" , "te" ,
 "sdr" , "gdf" , "df" , "fg" , "gh" , "oa" , "ah" , "qwe" , "re" , "ty" , "ui" };
        List<String> stringList = new ArrayList<>(Arrays.asList(strings)); 
 
        ForkJoinPool pool = new ForkJoinPool();
        ForkJoinService<List<String>> forkJoinService = ForkJoinTest.getInstance(stringList, 20 );
 // Submit a decomposable ForkJoinTask task  
        ForkJoinTask<List<String>> future = pool.submit(forkJoinService);
        System.out.println(future.get());
 // Close the thread pool  
 pool.shutdown(); 
 
 } 
 
 }