java 并发计算框架 forkjoin 使用

╰半橙微兮° 2023-07-23 13:56 56阅读 0赞

# **1.介绍** #

Fork/Join为JKD1.7引入，适用于对大量数据进行拆分成多个小任务进行计算的框架，

最后把所有小任务的结果汇总合并得到最终的结果

相关类

public abstract class RecursiveTask<V> extends ForkJoinTask<V>;
    public abstract class RecursiveAction extends ForkJoinTask<Void>;

其中RecursiveTask在执行有返回值的任务时使用，RecursiveAction在执行没有返回值的任务时使用

# 2.实例代码 #

## **ForkJoin有参无返回值** ##

public class CountRecursiveTask extends RecursiveAction {
        // 达到子任务直接计算的阈值
        private int num = 15;
    
        private int start;
        private int end;
    
        public CountRecursiveTask(int start, int end) {
            this.start = start;
            this.end = end;
        }
    
        @Override
        protected void compute() {
            if (this.end - this.start < num) {
                System.out.println("start:" + start + ";end:" + end);
            } else {
                //fork 2 tasks:Th = 15
                //如果仍大于阈值，则继续拆分为2个子任务，分别调用fork方法。
                //这里可以根据情况拆成n个子任务
                int middle = (end + start) / 2;
                CountRecursiveTask left = new CountRecursiveTask(start, middle);
                left.fork();
                System.out.println("start:" + start + ";middle:" + middle + ";end:" + end);
                CountRecursiveTask right = new CountRecursiveTask(middle + 1, end);
                right.fork();
    
            }
        }
    
        public static void main(String[] args){
            ForkJoinPool forkJoinPool = new ForkJoinPool(Runtime.getRuntime().availableProcessors());
    
            forkJoinPool.invoke(new CountRecursiveTask(1, 100));
        }
    
    }

## **ForkJoin有参有返回值(继承RecursiveTask<T>类)** ##

public class CountRecursiveTask extends RecursiveTask<Integer> {
        // 达到子任务直接计算的阈值
        private int num = 15;
    
        private int start;
        private int end;
    
        public CountRecursiveTask(int start, int end) {
            this.start = start;
            this.end = end;
        }
    
        @Override
        protected Integer compute() {
            if (this.end - this.start < num) {
                //如果小于阈值，直接调用最小任务的计算方法
                return count();
            } else {
                //fork 2 tasks:Th = 15
                //如果仍大于阈值，则继续拆分为2个子任务，分别调用fork方法。
                //这里可以根据情况拆成n个子任务
                int middle = (end + start) / 2;
                CountRecursiveTask left = new CountRecursiveTask(start, middle);
                System.out.println("start:" + start + ";middle:" + middle + ";end:" + end);
                left.fork();
                CountRecursiveTask right = new CountRecursiveTask(middle + 1, end);
                right.fork();
                //最后一定要记得fork()结果(如果需要结果的话)
                return left.join() + right.join();
            }
        }
    
        private int count() {
            int sum = 0;
            for (int i = start; i <= end; i++) {
                sum += i;
            }
            return sum;
        }
    
        public static void main(String[] args){
            ForkJoinPool forkJoinPool = new ForkJoinPool(Runtime.getRuntime().availableProcessors());
    
            Integer sum = forkJoinPool.invoke(new CountRecursiveTask(1, 100));
            System.out.println(sum);
        }
    
    }

在有大量计算任务时，此框架方法可进行并行计算效率高，

以上示例，可以根据具体的业务需求更改属性及相关方法用于匹配自己的业务逻辑

JDK1.8后由于加入Stream流的操作,集合框架可以使用

Collection<E> default Stream<E> parallelStream()的方法转换成并行流进行计算，此时效果与Fork/Join任务同效

ForkJoinPool中的多种方法

public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task);//等待获取结果
    public void execute(ForkJoinTask<?> task);//异步执行
    public <T> T invoke(ForkJoinTask<T> task);//执行,获取Future

ForkJoinTask在执行的时候可能会抛出异常，但是没办法在主线程里直接捕获异常，

所以ForkJoinTask提供了isCompletedAbnormally()方法来检查任务是否已经抛出异常或已经被取消了，

并且可以通过ForkJoinTask的getException方法获取异常。

getException方法返回Throwable对象，如果任务被取消了则返回CancellationException。

如果任务没有完成或者没有抛出异常则返回null。

if(task.isCompletedAbnormally()) {
        System.out.println(task.getException());
    }