Forkjoin -用来将大任务切分成多个小任务执行

前言

  如果您觉得有用的话，记得给博主点个赞，评论，收藏一键三连啊，写作不易啊^ _ ^。
  而且听说点赞的人每天的运气都不会太差，实在白嫖的话，那欢迎常来啊!!!

---

Forkjoin -用来将大任务切分成多个小任务执行

01 Forkjoin 是什么?

ForkJoinPool和ThreadPoolExecutor都是继承自AbstractExecutorService抽象类，所以它和ThreadPoolExecutor的使用几乎没有多少区别，
除了任务变成了ForkJoinTask以外。

注意的是ForkJoinPool使用无边界队列保存任务。 这些任务按线程池保护程序中指定的线程数执行。 如果未设置线程数，则默认情况下使用当前计算机上可用的CPU数或Docker容器中配置的CPU数大小的线程数。

02 ForkJoinPool和ThreadPoolExecutor的区别

ForkJoinPool和ThreadPoolExecutor最主要的区别就是ForkJoinPool中每一个线程都有属于自己的队列，当某个线程队列任务全部执行完了时，会通过"窃取工作"从别的线程队列中取出一个任务进行执行。

相较于线程池，Forkjoin池里的线程不是在是等待新任务，而是主动分裂的现有任务到更小的，并帮助完成其他线程的大任务(切分以后)。

03 使用

提交任务时，一般不会直接继承ForkJoinTask，只要继承它的子类即可。两者都提供了抽象方法compute，我们可以重写compute方法进行任务的划分。

• RecursiveAction:用于没有返回结果的任务(类似Runnable)
• RecursiveTask:用于有返回结果的任务(类似Callable)

04 示例

/**
* @author yangzhenyu
* @version 1.0
* @description:
* @date 2022/7/7 14:56
*/
public class TestVo implements  TestVos{
    private String name;
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

import com.alibaba.fastjson.JSONObject;
import com.yzy.vo.TestVo;
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;
import java.util.concurrent.RecursiveTask;

/**
* @author yangzhenyu
* @version 1.0
* @description:
* @date 2022/8/26 14:39
*/
public class ForkJoinPoolTest {
    private static class Task extends RecursiveTask<List<TestVo>> {


        private List<String> pkIds;


        public Task(List<String> pkIds) {
            this.pkIds = pkIds;
        }
        @Override
        protected List<TestVo> compute() {
            List<TestVo> result = new ArrayList<>();
            // 如果任务足够小就计算任务
            boolean canCompute = pkIds.size() <= 2;
            if (canCompute) {
                ArrayList<TestVo> objects = new ArrayList<>();
                //处理任务
                pkIds.forEach(v->{
                    TestVo testVo = new TestVo();
                    testVo.setName(v);
                    objects.add(testVo);
                });
                result = objects;
            } else {
                // 如果任务大于阈值，就分裂成两个子任务计算
                Task leftTask = new Task(pkIds.subList(0, pkIds.size() / 2));
                Task rightTask = new Task(pkIds.subList(pkIds.size() / 2, pkIds.size()));
                // 执行子任务
                leftTask.fork();
                rightTask.fork();
                // 等待子任务执行完，并得到其结果
                List<TestVo> left = leftTask.join();
                List<TestVo> right = rightTask.join();
                result.addAll(left);
                result.addAll(right);
            }
            return result;
        }
    }
    public static void main(String[] args) {
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        String [] pkIds =  new String []{"12","13","14","15","16","17"};
        Task task = new Task(Arrays.asList(pkIds));
        ForkJoinTask<List<TestVo>> result = forkJoinPool.submit(task);
        List<TestVo> list = new ArrayList<>();
        try {
            list = result.get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
            Thread.currentThread().interrupt();
        }
        System.out.println("return data 【list】:"+ JSONObject.toJSONString(list));


    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

效果: