Fork/Join使用学习

发表于

这是我参与11月更文挑战的第6天，活动详情查看：2021最后一次更文挑战。
JDK7提供了一个将任务“分而治之”的框架 — Fork/Join。它把一个大的任务分割成足够小的子任务，如果子任务比较大的话还要对子任务进行继续分割。分割的子任务分别放到双端队列里，然后启动线程分别从双端队列里获取任务执行。子任务执行完的结果都放在另外一个队列里，启动一个线程从队列里取数据，然后合并这些数据。

Fork/Join的思想如下所示：QQ截图20190702172347.png

RecursiveTask

QQ截图20190703111904.png

RecursiveTask适用于将任务分而治之，并且有返回值的情况，举个计算1到100和的例子：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
java复制代码public class RecursiveTest {
    // 定义最小区间为10
    private final static int MAX_THRESHOLD = 10;

    public static void main(String[] args) {
        final ForkJoinPool forkJoinPool = new ForkJoinPool();
        ForkJoinTask<Integer> future = forkJoinPool.submit(new CalculateRecursiveTask(1, 100));
        try {
            Integer result = future.get();
            System.out.println(result);
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }

    private static class CalculateRecursiveTask extends RecursiveTask<Integer> {
        // 起始
        private int start;
        // 结束
        private int end;

        public CalculateRecursiveTask(int start, int end) {
            this.start = start;
            this.end = end;
        }

        @Override
        protected Integer compute() {
            // 如果起始和结束范围小于我们定义的区间范围，则直接计算
            if ((end - start) <= MAX_THRESHOLD) {
                return IntStream.rangeClosed(start, end).sum();
            } else {
                // 否则，将范围一分为二，分成两个子任务
                int middle = (start + end) / 2;
                CalculateRecursiveTask leftTask = new CalculateRecursiveTask(start, middle);
                CalculateRecursiveTask rightTask = new CalculateRecursiveTask(middle + 1, end);
                // 执行子任务
                leftTask.fork();
                rightTask.fork();

                // 汇总子任务
                return leftTask.join() + rightTask.join();
            }
        }
    }
}

ForkJoinPool使用submit或invoke提交的区别：invoke是同步执行，调用之后需要等待任务完成，才能执行后面的代码；submit是异步执行，只有在Future调用get的时候会阻塞。

启动程序输出如下：

5050

其实这里执行子任务调用fork方法并不是最佳的选择，最佳的选择是invokeAll方法：

1
2
3
4
5
6
java复制代码// 执行子任务
// leftTask.fork();
// rightTask.fork();
invokeAll(leftTask,rightTask);
// 汇总子任务
return leftTask.join() + rightTask.join();

RecursiveAction

QQ截图20190703114411.png

使用方式和RecursiveTask类似，只不过没有返回值：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
java复制代码public class RecursiveActionTest {
    // 定义最小区间为10
    private final static int MAX_THRESHOLD = 10;
    private final static AtomicInteger SUM = new AtomicInteger(0);

    public static void main(String[] args) throws InterruptedException {
        final ForkJoinPool forkJoinPool = new ForkJoinPool();
        forkJoinPool.submit(new CalculateRecursiveAction(0, 100));
        forkJoinPool.awaitTermination(2, TimeUnit.SECONDS);
        System.out.println(SUM);
    }

    private static class CalculateRecursiveAction extends RecursiveAction {
        // 起始
        private final int start;
        // 结束
        private final int end;

        private CalculateRecursiveAction(int start, int end) {
            this.start = start;
            this.end = end;
        }

        @Override
        protected void compute() {
            // 如果起始和结束范围小于我们定义的区间范围，则直接计算
            if ((end - start) <= MAX_THRESHOLD) {
                SUM.addAndGet(IntStream.rangeClosed(start, end).sum());
            } else {
                // 否则，将范围一分为二，分成两个子任务
                int middle = (end + start) / 2;
                CalculateRecursiveAction leftAction = new CalculateRecursiveAction(start, middle);
                CalculateRecursiveAction rightAction = new CalculateRecursiveAction(middle + 1, end);
                // 执行子任务
                invokeAll(leftAction, rightAction);
                // 没有汇总子任务结果过程，因为没有返回值。
            }
        }
    }
}

输出结果也是5050。

理解Fork/Join框架API

Fork/Join框架在java.util.concurrent包下被实现。它的核心有4个类：

  • ForkJoinTask: 这是一个抽象任务类，并且运行在ForkJoinPool中。
  • ForkJoinPool:这是一个线程池管理并运行众多ForkJoinTask任务。
  • RecursiveAction: ForkJoinTask的子类，这个类没有返回值。
  • RecursiveTask: ForkJoinTask的子类，有返回值。

基本上，我们解决问题的代码是在RecursiveAction或者RecursiveTask中进行的，然后将任务提交由ForkJoinPool`执行，ForkJoinPool处理从线程管理到多核处理器的利用等各种事务。

我们先来理解一下这些类中的关键方法。

ForkJoinTask

这是一个运行在ForkJoinPool中的抽象的任务类。类型V指定了任务的返回结果。ForkJoinTask是一个类似线程的实体，它表示任务的轻量级抽象，而不是实际的执行线程。该机制允许由ForkJoinPool中的少量实际线程管理大量任务。其关键方法是：

  • final ForkJoinTask fork()
  • final V join()
  • final V invoke()

fork()方法提交并执行异步任务，该方法返回ForkJoinTask并且调用线程继续运行。

join()方法等待任务直到返回结果。

invoke()方法是组合了fork()和join()，它开始一个任务并等待结束返回结果。

此外，ForkJoinTask中还提供了用于一次调用多个任务的两个静态方法

  • static void invokeAll(ForkJoinTask task1, ForkJoinTask task2) :执行两个任务
  • static void invokeAll(ForkJoinTask<?>… taskList):执行任务集合

RecursiveAction

这是一个递归的ForkJoinTask子类，不返回结果。Recursive意思是任务可以通过分治策略分成自己的子任务（在下面的下一节中，您将看到如何划分代码示例）。

我们必须重写compute()方法,并将计算代码写在其中：

1
java复制代码protected abstract void compute();

RecursiveTask

和RecursiveAction一样，但是RecursiveTask有返回结果，结果类型由V指定。我们仍然需要重写compute()方法：

1
java复制代码protected abstract V compute();

ForkJoinPool

这是Fork/Join框架的核心类。它负责线程的管理和ForkJoinTask的执行，为了执行ForkJoinTask，首先需要获取到ForkJoinPool的实例。

有两种构造器方式可以获取ForkJoinPool的实例，第一种使用构造器创建：

  • ForkJoinPool(): 使用默认的构造器创建实例，该构造器创建出的池与系统中可用的处理器数量相等。
  • ForkJoinPool(int parallelism):该构造器指定处理器数量，创建具有自定义并行度级别的池，该级别的并行度必须大于0，且不超过可用处理器的实际数量。

并行性的级别决定了可以并发执行的线程的数量。换句话说，它决定了可以同时执行的任务的数量——但不能超过处理器的数量。

但是，这并不限制池可以管理的任务的数量。ForkJoinPool可以管理比其并行级别多得多的任务。

获取ForkJoinPool实例的第二种方法是使用以下ForkJoinPool的静态方法获取公共池实例:

1
java复制代码public static ForkJoinPool commonPool();

这种方式创建的池不受shutdown()或者shutdownNow()方法的影响，但是他会在System.exit()时会自动中止。任何依赖异步任务处理的程序在主体程序中止前都应该调用awaitQuiescence()方法。该方式是静态的，可以自动被使用。

作窃取算法

工作窃取（work-stealing）算法是指某个线程从其他队列里窃取任务来执行。

那么为什么需要使用工作窃取算法呢？假如我们需要做一个比较大的任务，我们可以把这个任务分割为若干互不依赖的子任务，为了减少线程间的竞争，于是把这些子任务分别放到不同的队列里，并为每个队列创建一个单独的线程来执行队列里的任务，线程和队列一一对应，比如A线程负责处理A队列里的任务。但是有的线程会先把自己队列里的任务干完，而其他线程对应的队列里还有任务等待处理。干完活的线程与其等着，不如去帮其他线程干活，于是它就去其他线程的队列里窃取一个任务来执行。而在这时它们会访问同一个队列，所以为了减少窃取任务线程和被窃取任务线程之间的竞争，通常会使用双端队列，被窃取任务线程永远从双端队列的头部拿任务执行，而窃取任务的线程永远从双端队列的尾部拿任务执行。

工作窃取算法的优点是充分利用线程进行并行计算，并减少了线程间的竞争，其缺点是在某些情况下还是存在竞争，比如双端队列里只有一个任务时。并且消耗了更多的系统资源，比如创建多个线程和多个双端队列。

什么时候用

上面只是为了演示Fork/Join的用法，实际是采用这种方式计算反而更加费时，因为切割任务，分配线程需要额外的开销。其实什么时候用不必太纠结，一个足够大的任务，如果采用Fork/Join来处理比传统处理方式快的话，那就毫不犹豫的选择它吧！

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%