Fork/Join框架

Fork-Join框架

分而治之:Fork/Join框架

分而治之是一种有效处理大量数据的方法，简单地说，如果你要处理1000个数据，但是又不具备处理它们的能力，那么你可以只处理其中的10个数据，再分阶段处理100次，将它们的结果进行合成，就是最终想要的结果。

 

在实际使用过程中，如果毫无顾忌的使用fork()方法开启线程进行处理，可能会导致线程过多而严重影响性能。所以在JDK中提供了一个ForkJoinPool线程池，对于fork（）方法不急于开启线程，而是提交给这个线程池进行处理，以节省系统资源。

 

由于线程池的优化，实际的任务数量和线程数量并不会是一一对应的关系，一个物理线程需要处理多个任务，因此也需要任务队列。实际处理过程中，可能会有这种情况:

A线程的任务已经处理完成，B线程堆积了许多任务，这时候A线程就会帮助B线程，从B线程的任务队列中获取任务来处理，尽可能的达到平衡，同时，处理自己的任务的时候，从任务队列的顶部开始获取数据，帮助别的线程的时候，从底部获取数据，尽量避免竞争。

 

看一下ForkJoinPool线程池的一个重要的方法:

通过submit方法可以向线程池提交一个ForkJoinTask任务，所谓ForkJoinTask任务就是支持fork()方法分解及join()方法等待的任务。ForkJoinTask有两个重要的子类:带返回值和不带返回值的两个抽象类。我们可以自定义任务，根据需要选择继承抽象类！

计算0加到90000000L

public class CountTask extends RecursiveTask<Long> {
    private static final int THRESHOLD = 10000;
    private long start;
    private long end;
​
    public CountTask(long start, long end) {
        this.start = start;
        this.end = end;
    }
​
    @Override
    protected Long compute() {
        long sum = 0;
        if (end - start <= 10000) {
            for (long i = start; i <= end; i++) {
                sum += i;
            }
        } else {
            //将任务先分成100个部分
            long step = (end - start) / 100;
            long pos = start;
            List<CountTask> taskList = new ArrayList<>();
            for (int i = 0; i < 100; i++) {
                long last = pos + step;
                if (last > end)
                    last = end;
                CountTask task = new CountTask(pos, last);
                task.fork();
                taskList.add(task);
                pos = last + 1;
            }
​
            for (CountTask task : taskList) {
                sum += task.join();
            }
        }
        return sum;
    }
​
    /**
     * 计算0加到90000000L
     * @param args
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        long start = System.currentTimeMillis();
        CountTask task = new CountTask(0, 90000000L);
        ForkJoinPool pool = new ForkJoinPool();
        ForkJoinTask<Long> result = pool.submit(task);
        long end = System.currentTimeMillis();
        System.out.println(end-start);
        System.out.println("result："+result.get());
    }
}

 

结果如下:

 

如果使用单线程累加计算:

public class CountDemo {
    public static void main(String[] args) {
        long start = System.currentTimeMillis();
        long sum = 0L;
        for (long i = 0; i <= 90000000L; i++) {
            sum += i;
        }
        long end = System.currentTimeMillis();
        System.out.println(end-start);
        System.out.println("result:"+sum);
    }
}

 

还是可以看到明显的差别的。

 

 

另外还有一点需要注意的是:fork方法会让子任务调用compute方法：

/**
 * Fork/Join框架
 */
public class ForkJoinDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ForkJoinPool pool = new ForkJoinPool();
        MyFork task = new MyFork(1, 100);
        ForkJoinTask<Integer> result = pool.submit(task);
        System.out.println(result.get());
        pool.shutdown();
    }
}
​
/**
 * 使用Fork/Join计算1加到100
 */
class MyFork extends RecursiveTask<Integer> {
    private static final Integer JUDGE_VALUE = 10;
    private int start;
    private int end;
​
    public MyFork(int start, int end) {
        this.start = start;
        this.end = end;
    }
​
    @Override
    protected Integer compute() {
        Integer result = 0;
        if (end - start <= JUDGE_VALUE) {
            for (int i = start; i <= end; i++) {
                result += i;
            }
        } else {
            System.out.println("fork");
            int mid = (end - start ) / 2 + start;
            MyFork task1 = new MyFork(start, mid);
            MyFork task2 = new MyFork(mid + 1, end);
            task1.fork();
            task2.fork();
            result = task1.join() + task2.join();
        }
        return result;
    }
}

虽然在compute中只分成2个子任务，但是最终fork打印了15次，说明task1和task2实际上还被划分成了多个子任务执行。