Java多线程再学习，温故知新（十七）ForkJoin框架详解

￡神魔★判官ぃ 2022-03-20 01:42 226阅读 0赞

ForkJoin是Java7提供的原生多线程并行处理框架，其基本思想是将大任物分割成小任务，最后将小任务聚合起来得到结果。它非常类似于HADOOP提供的MapReduce框架，只是MapReduce的任务可以针对集群内的所有计算节点，可以充分利用集群的能力完成计算任务。ForkJoin更加类似于单机版的MapReduce。多线程的目的不仅仅是提高程序运行的性能，但是可以充分利用CPU资源

Fork/Join使用两个类完成以上两件事情：

ForkJoinTask: 我们要使用ForkJoin框架，必须首先创建一个ForkJoin任务。它提供在任务中执行fork()和join的操作机制，通常我们不直接继承ForkjoinTask类，只需要直接继承其子类。  
   
    1. RecursiveAction，用于没有返回结果的任务  
    2. RecursiveTask，用于有返回值的任务  
   
ForkJoinPool：task要通过ForkJoinPool来执行，分割的子任务也会添加到当前工作线程的双端队列中，进入队列的头部。当一个工作线程中没有任务时，会从其他工作线程的队列尾部获取一个任务。  
   
ForkJoin框架使用了工作窃取的思想（work-stealing），算法从其他队列中窃取任务来执行  
  
简单实例

import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.Future;
    import java.util.concurrent.RecursiveTask;
    
    public class Demo extends RecursiveTask<Integer> {
    
    	private int begin;
    	private int end;
    
    	public Demo(int begin, int end) {
    		this.begin = begin;
    		this.end = end;
    	}
    
    	@Override
    	protected Integer compute() {
    		System.out.println(Thread.currentThread().getName() + " ... ");
    		
    		int sum = 0;
    		// 拆分任务
    		if (end - begin <= 2) {
    			// 计算
    			for (int i = begin; i <= end; i++) {
    				sum += i;
    			}
    		} else {
    			// 拆分
    			
    			Demo d1 = new Demo(begin, (begin + end) / 2);
    			Demo d2 = new Demo((begin + end)/2 + 1, end);
    			
    			// 执行任务
    			d1.fork();
    			d2.fork();
    			
    			Integer a = d1.join();
    			Integer b = d2.join();
    			
    			sum = a + b;
    		}
    
    		return sum;
    	}
    
    	public static void main(String[] args) throws Exception {
    
    		ForkJoinPool pool = new ForkJoinPool(3);
    
    		Future<Integer> future = pool.submit(new Demo(1, 1000000000));
    
    		System.out.println("....");
    
    		System.out.println("计算的值为：" + future.get());
    	}
    
    }