23、详解java中一个分而治之的框架ForkJoin

我就是我 2023-05-30 08:53 4阅读 0赞

在古代，皇帝要想办成一件事肯定不会自己亲自去动手，而是把任务细分发给下面的大臣，下面的大臣也懒呀，于是把任务继续分成几个部分，继续下发，于是到了最后最终负责的人就完成了一个小功能。上面的领导再把这些结果一层一层汇总，最终返回给皇帝。这就是分而治之的思想，也是我们今天的主题ForkJoin。

一、简介

从JDK1.7开始，Java提供ForkJoin框架用于并行执行任务，它的思想就是讲一个大任务分割成若干小任务，最终汇总每个小任务的结果得到这个大任务的结果。

**1、ForkJoinPool**

既然任务是被逐渐的细化的，那就需要把这些任务存在一个池子里面，这个池子就是ForkJoinPool，它与其它的ExecutorService区别主要在于它使用“工作窃取“，那什么是工作窃取呢？

> **一个大任务会被划分成无数个小任务，这些任务被分配到不同的队列，这些队列有些干活干的块，有些干得慢。于是干的快的，一看自己没任务需要执行了，就去隔壁的队列里面拿去任务执行。**

**2、ForkJoinTask**

ForkJoinTask就是ForkJoinPool里面的每一个任务。他主要有两个子类：RecursiveAction和RecursiveTask。然后通过fork()方法去分配任务执行任务，通过join()方法汇总任务结果，

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NERERMTEw_size_16_color_FFFFFF_t_70]

这就是整个过程的运用。他有两个子类，使用这两个子类都可以实现我们的任务分配和计算。

（1）RecursiveAction 一个递归无结果的ForkJoinTask（没有返回值）

（2）RecursiveTask 一个递归有结果的ForkJoinTask（有返回值）

ForkJoinPool由ForkJoinTask数组和ForkJoinWorkerThread数组组成，ForkJoinTask数组负责存放程序提交给ForkJoinPool的任务，而ForkJoinWorkerThread数组负责执行这些任务。

下面我们就来看看如何去使用。

二、使用

1、RecursiveTask ：有返回结果

（1）第一步：创建MyRecursiveTask子类在ForkJoinTest中

private static class MyRecursiveTask extends RecursiveTask<Integer>{ 
    		private final int start;
    		private final int end;
    		public MyRecursiveTask(int start, int end) { 
    			this.start = start;
    			this.end = end;
    		}
    		@Override
    		protected Integer compute() { 
    			//如果任务小的不能再细分了，那就直接计算
    			if(end-start<=threshold) { 
    				//rangeClosed包含结束节点，是闭合的
    				return IntStream.rangeClosed(start,end).sum();
    			}else { 
    				int mid = (start + end ) /2;
    				MyRecursiveTask leftTask = new MyRecursiveTask(start, mid);
    				MyRecursiveTask rightTask = new MyRecursiveTask(mid+1, end);
    				leftTask.fork();
    				rightTask.fork();
    				return leftTask.join() + rightTask.join() ;
    			}	
    		}
    	}

在这个方法中我们传进去数据，然后使用二分法继续分配给子任务，当任务小的不能再分，那就汇总返回。

（2）第二步在ForkJoinTest中去测试

public class ForkJoinTest { 
    	private final static int threshold = 3;
    	public static void main(String[] args) { 
    		final ForkJoinPool pool = new ForkJoinPool();
    		ForkJoinTask<Integer> result = pool.submit(new MyRecursiveTask(0, 100));
    		try { 
    			System.out.println(result.get());
    		} catch (Exception e) { 
    			e.printStackTrace();
    		}
    	}
    }

在这个类中我们定义了一个阈值，然后创建一个ForkJoinPool，在这个池子中新建我们刚刚创建的Task任务，最终返回我们结果。

2、RecursiveAction：无返回结果

（1）第一步：创建MyRecursiveAction子类在ForkJoinTest中

private static class MyRecursiveAction extends RecursiveAction{ 
    		private final int start;
    		private final int end;	
    		public MyRecursiveAction(int start, int end) { 
    			this.start = start;
    			this.end = end;
    		}
    		@Override
    		protected void compute() { 
    			//如果任务小的不能再细分了，那就直接计算
    			if(end-start<=threshold) { 
    				//rangeClosed包含结束节点，是闭合的
    				sum.addAndGet(IntStream.rangeClosed(start, end).sum());
    			}else { 
    				int mid = (start + end ) /2;
    				MyRecursiveTask leftTask = new MyRecursiveTask(start, mid);
    				MyRecursiveTask rightTask = new MyRecursiveTask(mid+1, end);
    				leftTask.fork();
    				rightTask.fork();
    			}
    		}
    	}

在这个方法中我们不需要有return语句。过程和之前的RecursiveTask类似。

（2）第二步在ForkJoinTest中去测试

public class ForkJoinTest { 
    	private final static int threshold = 3;
    	private final static AtomicInteger sum = new AtomicInteger(0);
    	public static void main(String[] args) throws InterruptedException { 
    		final ForkJoinPool pool = new ForkJoinPool();
            //注意在这里不需要有返回值了
    		pool.submit(new MyRecursiveAction(0, 100));
    		pool.awaitTermination(100, TimeUnit.MILLISECONDS);
            //我们依然还可以输出最终的返回值
            System.out.println(sum);
    	}
    }

现在不管我们输出多少次都可以有返回结果了。

ForkJoinTask在执行的时候可能会抛出异常，在主线程中是无法直接获取的，但是可以通过ForkJoinTask提供的isCompletedAbnormally()方法来检查任务是否已经抛出异常或已经被取消了。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NERERMTEw_size_16_color_FFFFFF_t_70 1]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NERERMTEw_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191108155951817.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NERERMTEw=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NERERMTEw_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20191108160014221.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NERERMTEw=,size_16,color_FFFFFF,t_70