面试官: Fork/Join 有了解过吗？说说看（含源码分析）

2023-08-07 16:52| 来源: 网络整理| 查看: 265

前言

目前正在出一个Java多线程专题长期系列教程，从入门到进阶含源码解读, 篇幅会较多, 喜欢的话，给个关注❤️ ~

本节给大家介绍一个同样实现了ExecutorService接口的多线程处理器Fork/Join框架,一起来看下吧~

Fork/Join

首先从字面意思讲,fork有分发的意思，join有聚合的意思，所以大胆猜测一下，fork就是把一个大的任务拆分成诸多小的任务，join就是将最终的结果结合起来。

下面我们通过一个累加的demo快速的认识一下它，这里尽量的简单，让大家好理解 ~

public class ForkJoinTest { // 首先继承 RecursiveTask 并重写compute方法 public static class TaskDemo extends RecursiveTask { private Integer i = 0; private Integer num; public TaskDemo(Integer num) { this.num = num; } @Override protected Integer compute() { // 判断任务是否拆分到合适数量 if(num > 20) { // 任务拆分完成后进行计算 for(int index=0; index < 200; index++) { i ++; } return i; }else { // 拆分成两个任务 TaskDemo t1 = new TaskDemo(num + 1); t1.fork(); TaskDemo t2 = new TaskDemo(num + 2); t2.fork(); return t1.join() + t2.join(); } } } public static void main(String[] args) throws ExecutionException, InterruptedException { Long start = System.currentTimeMillis(); // 创建线程池 ForkJoinPool pool = new ForkJoinPool(); // 捕获返回的最终结果 ForkJoinTask taskFuture = pool.submit(new TaskDemo(0)); Integer result = taskFuture.get(); System.out.println("最终结果 >>>> " + result); Long end = System.currentTimeMillis(); System.out.println("耗时 >>> " + (end -start) + "ms"); } } 复制代码

实际输出

最终结果 >>>> 5731400 耗时 >>> 170ms 复制代码

可以看到，执行速度非常的快，它可以发挥cpu最大性能，任务越大越明显。从上面的用法来看，Fork/Join框架简单来说就是切割任务与子任务进行合并，最终返回结果。

知其然知其所以然 & 源码分析

下面就带大家简单的看一下它的源码实现,我们重点看以下几个方法:

fork()

方法: 使用线程池中的空闲线程异步提交任务

public final ForkJoinTask fork() { Thread t; // ForkJoinWorkerThread是执行ForkJoinTask的专有线程，由ForkJoinPool管理 // 先判断当前线程是否是ForkJoin专有线程，如果是，则将任务push到当前线程所负责的队列里去 if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ((ForkJoinWorkerThread)t).workQueue.push(this); else // 如果不是则将线程加入队列 // 没有显式创建ForkJoinPool的时候走这里，提交任务到默认的common线程池中,之前的例子是手动创建了一个 ForkJoinPool.common.externalPush(this); return this; } 复制代码

这个方法只做一件事情, 把任务推入当前工作线程的工作队列里。

join()

方法: 等待处理任务的线程处理完毕，获得返回值。

public final V join() { int s; // doJoin()来获取当前任务的执行状态 if ((s = doJoin() & DONE_MASK) != NORMAL) reportException(s); // 最后返回结果 return getRawResult(); } 复制代码

doJoin() 方法用来返回当前任务的执行状态

private int doJoin() { int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w; // 先判断任务是否执行完毕，执行完毕直接返回结果（执行状态） return (s = status) < 0 ? s : // 如果没有执行完毕，先判断是否是ForkJoinWorkThread线程 ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ? // 如果是，先判断任务是否处于工作队列顶端（意味着下一个就执行它） // tryUnpush()方法判断任务是否处于当前工作队列顶端，是返回true // doExec()方法执行任务 (w = (wt = (ForkJoinWorkerThread)t).workQueue). // 如果是处于顶端并且任务执行完毕，返回结果 tryUnpush(this) && (s = doExec()) < 0 ? s : // 如果不在顶端或者在顶端却没执行完毕，那就调用awitJoin()执行任务 // awaitJoin()：使用自旋使任务执行完成，返回结果 wt.pool.awaitJoin(w, this, 0L) : // 如果不是ForkJoinWorkThread线程，执行externalAwaitDone()返回任务结果 externalAwaitDone(); } 复制代码

工作窃取算法

Fork/Join框架在处理任务时使用了工作窃取算法, 工作窃取算法指的是在多线程执行不同任务队列的过程中，某个线程执行完自己队列的任务后从其他线程的任务队列里窃取任务来执行。

当一个线程窃取另一个线程的时候，为了减少两个任务线程之间的竞争，我们通常使用双端队列来存储任务。被窃取的任务线程都从双端队列的头部进行任务执行，而窃取其他任务的线程从双端队列的尾部执行任务。

另外，当一个线程在窃取任务时没有可用任务，就会进入阻塞状态以等待再次工作。

结束语

本节主要讲解它的一个使用，有兴趣的同学可以继续看一下它的底层源码实现。下一节，给大家讲下Stream流式计算多线程处理，关注我，不迷路 ~

【本文地址】

公司简介

联系我们