JAVA8-Fork Join

Fork：将一个任务拆分成多个线程执行。
Join：将每个现场结果join，最后得到结果。

范例：求数组总和

前置数据：

public static int[] data = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10};

1、原始实现方式

    public static int calc() {
        int result = 0;
        for (int i = 0; i < data.length; i++) {
            result += data[i];
        }
        return result;
    }

原始实现方式调用：

System.out.println("result=> " + calc());

2、RecursiveTask实现

RecursiveTask有返回值。

package com.example.study.java8.forkjoin;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.ToString;

import java.util.concurrent.RecursiveTask;
@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString
public class AccumulatorRecursiveTask extends RecursiveTask<Integer> {
    private  int start;
    private  int end;
    private  int[] data;
    private  int LIMIT=3;

    public AccumulatorRecursiveTask(int start, int end, int[] data) {
        this.start = start;
        this.end = end;
        this.data = data;
    }

    @Override
    protected Integer compute() {
        //如果小于3个就可以直接进行计算了，否则fork成多个，再计算
        if(end -start <LIMIT) {
            int result =0;
            for(int i=start;i<end;i++){
                result+=data[i];
            }
            return result;
        }

        int mid = (start+end)/2;
        AccumulatorRecursiveTask left = new AccumulatorRecursiveTask(start,mid,data);
        AccumulatorRecursiveTask right = new AccumulatorRecursiveTask(mid,end,data);

        left.fork();
        Integer rightResult = right.compute();
        Integer leftResult = left.join();
        return rightResult+leftResult;
    }
}

RecursiveTask调用：

        AccumulatorRecursiveTask task = new AccumulatorRecursiveTask(0, data.length, data);
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        Integer result = forkJoinPool.invoke(task);
        System.out.println("AccumulatorRecursiveTask result => " + result);

3、RecursiveAction实现

RecursiveAction:无返回值

package com.example.study.java8.forkjoin;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.ToString;

import java.util.concurrent.RecursiveAction;
import java.util.concurrent.atomic.AtomicInteger;
@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString
public class AccumulatorRecursiveAction extends RecursiveAction {
    private int start;
    private int end;
    private int[] data;
    private int LIMIT = 3;

    public AccumulatorRecursiveAction(int start, int end, int[] data) {
        this.start = start;
        this.end = end;
        this.data = data;
    }

    @Override
    protected void compute() {
        //如果小于3个就可以直接进行计算了，否则fork成多个，再计算
        if (end - start < LIMIT) {
            for (int i = start; i < end; i++) {
                AccumulatorHelper.accumulate(data[i]);
            }
        } else {
            int mid = (start + end) / 2;
            AccumulatorRecursiveAction left = new AccumulatorRecursiveAction(start, mid, data);
            AccumulatorRecursiveAction right = new AccumulatorRecursiveAction(mid, end, data);
            right.fork();
            left.fork();
            right.join();
            left.join();
        }
    }

    static class AccumulatorHelper {
        private static final AtomicInteger result = new AtomicInteger(0);

        static void accumulate(int value) {
            result.getAndAdd(value);
        }

        public static int getResult() {
            return result.get();
        }

        static void rest() {
            result.set(0);
        }
    }
}

RecursiveAction调用：

        AccumulatorRecursiveAction action = new AccumulatorRecursiveAction(0, data.length, data);
        forkJoinPool.invoke(action);
        System.out.println("AccumulatorRecursiveAction result => " + AccumulatorRecursiveAction.AccumulatorHelper.getResult());

最后输出结果

result=> 55
AccumulatorRecursiveTask result => 55
AccumulatorRecursiveAction result => 55