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

Posted 2022-12-28 栗子~~

文章目录

• 前言
• • Forkjoin -用来将大任务切分成多个小任务执行
  • • 01 Forkjoin 是什么?
    • 02 ForkJoinPool和ThreadPoolExecutor的区别
    • 03 使用
    • 04 示例

前言

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

---

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;
    

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));


    

效果: