ForkJoin框架从入门到使用

╰+攻爆jí腚メ 2024-04-17 06:23 77阅读 0赞

先附上中文版API链接  
https://ifeve.com/tag/forkjoin/

# ForkJoin: #

体现了分而治之的思想，大于既定大小，分为很多子任务，小于既定大小，直接调用任务解决  
（自己判断情形），子任务是相对独立的，与原问题形式相同，将子问题的解合并得到原问题的解（有点类似mapReduce）  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzAwMDMw_size_16_color_FFFFFF_t_70]  
fork操作：当你把任务分成更小的任务和使用这个框架执行它们。  
join操作：等待子任务线程执行完毕，返回最终结果。

，Fork/Join框架执行的任务有以下**局限性**：

1、任务只能使用fork()和join()操作，作为同步机制。如果使用其他同步机制，工作线程不能执行其他任务，当它们在同步操作时。比如，在Fork/Join框架中，你使任务进入睡眠，正在执行这个任务的工作线程将不会执行其他任务，在这睡眠期间内。  
2、任务不应该执行I/O操作，如读或写数据文件。  
3、任务不能抛出检查异常，它必须包括必要的代码来处理它们。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzAwMDMw_size_16_color_FFFFFF_t_70 1]  
核心类：  
**ForkJoinPool**：它实现ExecutorService接口和work-stealing（工作密取）算法。它管理工作线程和提供关于任务的状态和它们执行的信息。  
**ForkJoinTask**： 它是将在ForkJoinPool中执行的任务的基类。它提供在任务中执行fork()和join()操作的机制，并且这两个方法控制任务的状态。通常， 为了实现你的Fork/Join任务，你将实现两个子类的子类的类：**RecursiveAction**对于没有返回结果的任务，**RecursiveTask** 对于返回结果的任务。

**同步和异步执行ForkJoin:** 同步模式时（pool.invoke()），只有forkJoin的任务执行完成，主线程才继续往下执行。异步时( pool.execute)，不管forkJoin是否执行完成，主线程依然往下执行

注意几点：RecursiveTask/RecursiveAction集成时需要指定返回类型，通过**invokeAll**方法将子任务传入任务队列，递归需要有出口，多个子任务时需要将子任务的值处理成一个后返回；

演示：Fork/Join的同步用法同时演示返回结果值：统计整形数组中所有元素的和

private static class SumTask extends RecursiveTask<Integer>{
            //MakeArray.ARRAY_LENGTH代表数组的长度
            private final static int THRESHOLD = MakeArray.ARRAY_LENGTH/10;//阈值，
            private int[] src; //表示我们要实际统计的数组
            private int fromIndex;//开始统计的下标
            private int toIndex;//统计到哪里结束的下标
    
            public SumTask(int[] src, int fromIndex, int toIndex) {
                this.src = src;
                this.fromIndex = fromIndex;
                this.toIndex = toIndex;
            }
    
    		@Override
    		protected Integer compute() {
                //当数组的大小阈值已经不需要继续划分子任务，执行业务操作
    			if(toIndex-fromIndex < THRESHOLD) {
    				int count = 0;
    				for(int i=fromIndex;i<=toIndex;i++) {
    			    	count = count + src[i];
    				}
    				return count;
    			}else {//任务过大，继续划分子任务
    				int mid = (fromIndex+toIndex)/2;
    				SumTask left = new SumTask(src,fromIndex,mid);
    				SumTask right = new SumTask(src,mid+1,toIndex);
    				invokeAll(left,right);//将划分的子任务传入新的队列
    				return left.join()+right.join();
    			}
    		}
    		public static void main(String[] args) {
        ForkJoinPool pool = new ForkJoinPool();//创建任务池
        int[] src = MakeAarry.makeArray();//自定义方法，创建一个数组
        SumTask innerFind = new SumTask(src, 0, src.length - 1);
        pool.invoke(innerFind);//同步调用，只有执行完这一步之后，主线程才继续往下执行
    }
        }

再简单实现一个异步读取目录结构的方法；

//一个针对文件的异步读取目录方法
    public class FileForkJoin extends RecursiveAction {
        private File path;//当前任务需要搜索的目录
        public FileForkJoin(File path) {
            this.path = path;
        }
        @Override
        protected void compute() {//重写这个方法
            List<FileForkJoin> tasks = new ArrayList<>();//需要被执行的子任务集合
            File[] files = path.listFiles();//找出此目录下的内容
            if (files != null) {
                for (File file : files) {
                    if (file.isDirectory()) {
                        tasks.add(new FileForkJoin(file));
                    } else {
                        if (file.getAbsolutePath().endsWith(".jpg")) {
                            System.out.println("文件：" + file.getAbsolutePath());
                        }
                    }
                }
                if (!tasks.isEmpty()) {
                    invokeAll(tasks);//添加任务集合，返回值就是tasks
                    tasks.forEach(t -> t.join());//挨个添加任务
                }
            }
    
        }
        public static void main(String[] args) {
            ForkJoinPool pool = new ForkJoinPool();
            File path = new File("D:/");
            FileForkJoin fileForkJoin = new FileForkJoin(path);
            pool.execute(fileForkJoin);//异步执行
            fileForkJoin.join();//阻塞
        }
    }

# 工作密取 #

工作窃取算法是指某个线程从其他队列里窃取任务来执行。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzAwMDMw_size_16_color_FFFFFF_t_70 2]  
**那么，为什么需要使用工作窃取算法呢？**  
假如我们需要做一个比较大的任务，可以把这个任务分割为若干互不依赖的子任务，为了减少线程间的竞争，把这些子任务分别放到不同的队列里，并为每个队列创建一个单独的线程来执行队列里的任务，线程和队列一一对应。比如A线程负责处理A队列里的任务。但是，有的线程会先把自己队列里的任务干完，而其他线程对应的队列里还有任务等待处理。干完活的线程与其等着，不如去帮其他线程干活，于是他就去其他线程的队列里窃取一个任务来执行。而在这时他们会访问同一个队列，所以为了减少窃取任务线程和被窃取任务线程之间的竞争，通常会使用双端队列，被窃取任务线程永远从双端队列的头部拿任务执行，而窃取任务的线程永远从双端队列的尾部拿任务执行。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzAwMDMw_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/c128d9f4e39d473db7c7d6262470b1bf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzAwMDMw_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/c42202b0c6c74330b129036035b33398.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzAwMDMw_size_16_color_FFFFFF_t_70 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/c946005e2cb049689ef6b62ce503e6fe.png