java并发编程之Fork/Join框架

女爷i 2023-06-27 08:53 5阅读 0赞

Fork/Join是java提供的一个用于并行执行任务的框架，通过把一个大的任务分成若干个小任务，最后再把每个小任务的结果汇总得到这个大任务的结果。

举个例子，比如你要计算1+2+3+4+5+6的和，可以把这个任务分为2个小任务，小任务1负责计算1+2+3的和，小任务2负责计算4+5+6的和，然后再把这2个小任务的和汇总就是1+2+3+4+5+6的和了。

我们通过一个运行流程图来看下吧。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6czUzNTEzMQ_size_16_color_FFFFFF_t_70][]

通过这个图可以看到，Fork就是把一个大的任务分成几个较小的任务去执行，join就是把几个小任务的结果合并最终形成大任务的结果，这就是Fork/Join框架名字的由来。

讲到Fork/Join框架就不得不说一个算法，那就是工作窃取算法。

工作窃取算法顾名思义就是窃取工作的算法，就是指某个线程从其它的队列中窃取任务来执行。

一般来说，我们执行一个大的任务，就会把这个任务分成若干个小任务来执行，这些个小任务是互不依赖的，为了减少线程之间的竞争，就会把这些的小任务放到不同的队列中来执行，然后为每一个队列创建一个线程执行这个队列中的任务，比如Thread1执行队列1中的任务，Thread2执行队列2中的任务，但是如果Thread2执行完了队列1中的任务，Thread2还没有执行完队列2中的任务，怎么办呢？如果Thread1如果什么也不做等待Thread2的完成，还不如去帮助Thread2去执行队列2中的任务呢，这样的话效率更高些。但是如果Thread1和Thread2同时访问同一个队列，容易产生竞争，因此要使用双端队列，这样的话，线程执行自己的队列就从队列头部拿任务，窃取线程就从队列尾部拿任务。

如图所示，工作窃取流程图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6czUzNTEzMQ_size_16_color_FFFFFF_t_70 1][]

Fork/Join框架中主要有2个类来完成这些事情。

一、ForkJoinTask ：我们要使用这个框架就要创建个ForkJoin任务，在ForkJoin中执行fork()和join()的操作，一般我们继承ForkJoinTask的子类就好了，它有2个子类。

（1）RecurisiveAction：用于不返回结果的任务。

（2）RecurisiveTask：用于有返回结果的任务。

二、ForkJoinPool：ForkJoinTask需要ForkJoinPool来执行。

如果不理解的话，我们来看下代码吧。

package com.chongzi.keji.other;
    
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.Future;
    import java.util.concurrent.RecursiveTask;
    
    public class SumTask extends RecursiveTask<Integer>{//需要返回结果所以继承RecursiveTask
    
    	private static final int Threshold = 2;
    	private int start;
    	private int end;
    	
    	public SumTask(int start,int end) {
    		this.start = start;
    		this.end = end;
    	}
    	
    
    	@Override
    	protected Integer compute() {
    		int sum = 0;
    		boolean isCompute = (end - start)<=Threshold;
    		if(isCompute){
    			for(int j = start;j<=end;j++){
    				sum = sum + j;
    			}
    		}else{
    			int mid = (start + end)/2;
    			SumTask task1 = new SumTask(start, mid);
    			SumTask task2 = new SumTask(mid+1, end);
    			task1.fork();
    			task2.fork();
    			int result1 = task1.join();
    			int result2 = task2.join();
    			sum = result1+ result2;
    		}
    		return sum;
    	}
    
        public static void main(String[] args) {
    		ForkJoinPool forkJoinPool = new ForkJoinPool();
    		//我们计算1+2+3+4=5+6的和
    		SumTask task = new SumTask(1, 6);
    		Future<Integer> future = forkJoinPool.submit(task);
    		try {
    			System.out.print(future.get());
    		} catch (InterruptedException e) {
    			
    			e.printStackTrace();
    		} catch (ExecutionException e) {
    			
    			e.printStackTrace();
    		}
    
    	}
    }

通过这个例子我们可以看到compute方法必须要实现，主要的逻辑也在这个方法中，通过fork方法执行子任务后，还要执行join方法等待子任务的完成并且得到子任务的结果，然后把汇总的结果返回，就是最终的任务的结果。

Fork/Join框架你会用了吗？学到的点个赞呗。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6czUzNTEzMQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200105220716451.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6czUzNTEzMQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6czUzNTEzMQ_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200105220812321.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d6czUzNTEzMQ==,size_16,color_FFFFFF,t_70