Source code of this article: GitHub · click here | gitee · click here

1、 Fork / join framework

Java provides fork / join framework for parallel execution of tasks. The core idea is to divide a large task into multiple small tasks, and then summarize the execution results of each small task to get the final result of the large task.

This mechanism strategy is very common in distributed databases. The data is distributed in replicas of different databases. When executing queries, each service has to run query tasks. Finally, data consolidation is performed on one service, or an intermediate engine layer is provided to summarize the total data:

Core processes: task segmentation, asynchronous execution of module tasks, and consolidation of single task results; In programming, there are few common codes, but common ideas can be seen everywhere.

2、 Core APIs and methods

1. Coding case

Based on 1 + 2+ A calculation case of 100 demonstrates the basic usage of fork / join framework.

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;

public class ForkJoin01 {

    public static void main (String[] args) {
        int[] numArr = new int[100];
        for (int i = 0; i < 100; i++) {
            numArr[i] = i + 1;
        }
        ForkJoinPool pool = new ForkJoinPool();
        ForkJoinTask<Integer> forkJoinTask =
                pool.submit(new SumTask(numArr,numArr.length));
        System.out.println("合并计算结果: " + forkJoinTask.invoke());
        pool.shutdown();
    }
}
/**
 * 线程任务
 */
class SumTask extends RecursiveTask<Integer> {
    /*
     * 切分任务块的阈值
     * 如果THRESHOLD=100
     * 输出：main【求和：(0...100)=5050】 合并计算结果: 5050
     */
    private static final int THRESHOLD = 100;
    private int arr[];
    private int start;
    private int over;

    public SumTask(int[] arr,int start,int over) {
        this.arr = arr;
        this.start = start;
        this.over = over;
    }

    // 求和计算
    private Integer sumCalculate () {
        Integer sum = 0;
        for (int i = start; i < over; i++) {
            sum += arr[i];
        }
        String task = "【求和：(" + start + "..." + over + ")=" + sum +"】";
        System.out.println(Thread.currentThread().getName() + task);
        return sum ;
    }

    @Override
    protected Integer compute() {
        if ((over - start) <= THRESHOLD) {
            return sumCalculate();
        }else {
            int middle = (start + over) / 2;
            SumTask left = new SumTask(arr,start,middle);
            SumTask right = new SumTask(arr,middle,over);
            left.fork();
            right.fork();
            return left.join() + right.join();
        }
    }
}

2. Core API description

Forkjoinpool: the biggest feature of thread pool is the fork join mode, which splits a large task into multiple small tasks for parallel execution, and then combines the work stealing algorithm to improve the overall execution efficiency and make full use of CPU resources.

Forkjointask: a task abstraction running in the forkjoinpool, which can be understood as a thread like entity but is lighter than a thread. A small number of forkjoinworkerthreads running in the forkjoinpool can hold a large number of forkjointasks and its subtasks. At the same time, it is also a lightweight future. Long blocking or IO should be avoided in use.

Inherit subclasses:

Core approach:

3. Core strategy description

Task splitting

Based on the divide and conquer algorithm, forkjoinpool continuously splits large tasks, and splits each sub task in half until the task granularity set by the most threshold is reached. It puts the tasks into different queues, then performs calculations from the lowest task, and merges the results to the upper layer. In this way, a large number of tasks are processed with relatively few threads.

Job theft algorithm

Large tasks are divided into independent subtasks, and subtasks are placed in different queues, and a thread is created for each queue to execute the tasks in the queue. Assuming that thread a gives priority to executing the tasks assigned to its queue, if there are tasks waiting to be executed in the queue corresponding to thread e, idle thread a will steal the tasks in thread e's queue for execution, In addition, in order to reduce the competition between thread a and stolen task thread e when stealing a task, thread a stealing a task will obtain task execution from the tail of the queue, and the stolen task thread e will obtain task execution from the head of the queue.

Advantages of work stealing algorithm: there is little competition among threads. Threads are fully used for parallel computing, but competition may also exist when there is only one task in the task queue.

3、 Application case analysis

In the business development of the back-end system, it can be used for various functional scenarios such as permission verification, batch timing, task status refresh and so on:

As shown in the figure above, assuming that the primary key ID of the data is segmented as follows, the data scenario may be the connection information of the data source, or similar businesses with product validity, which can be processed based on thread pool tasks:

Permission verification

Judge whether the data source is available based on the connection information of the data source. For example, judge whether the connection is available and whether the user has the read-write permission of the library table. When there are many data sources, check quickly based on the thread pool.

Status refresh

In scheduled tasks, you often see status class refresh operations, such as judging whether the product is within the validity period, and setting the data to the invalid state outside the validity period. You can use the thread pool for rapid processing.

4、 Source code address

GitHub·地址
https://github.com/cicadasmile/java-base-parent
GitEE·地址
https://gitee.com/cicadasmile/java-base-parent