概述

ForkJoinPool分支/合并框架，就是在必要的情况下，将一个大任务拆分（fork）成若干个小任务（拆到不能再拆为止），再将一个个的小任务运算的结果进行Join汇总。

ThreadPool与ForkJoinPool介绍

ThreadPool Executor

一个线程池包括以下四个基本组成部分：

线程管理器（ThreadPool）：用于创建并管理线程池，包括创建线程池，销毁线程池，添加新任务。
工作线程（PoolWorker）：线程池中的线程，在没有任务时处于等待状态，可以循环的执行任务；
任务接口（Task）：每个任务必须实现的接口，以供工作线程调度任务的执行，它注意规定了任务的入口，任务执行完后的收尾工作，任务的执行状态等；
任务队列（taskQueue）：用于存放没有处理的任务，提供一种缓冲机制。

工作方式

线程池有一个工作队列，队列中包含了要分配给各线程的工作（任务）。当线程空闲时，就会从队列中认领工作。由于线程资源的创建于销毁开销很大，所以ThreadPool允许线程的重用，减少创建于销毁的次数，提供效率。

流程细节

ForkJoinPool Executor

ForkJoinPool组成类

ForkJoinPool：充当fork/join框架里面的管理者，最原始的任务都要交给它才能处理。它负责控制整个fork/join有多少workerThread；workerThread的创建，激活都是由它来掌控的。它还负责workQueue队列的创建和分配；每当创建一个workerThread，它负责分配相应的workQueue，然后它把接到的活都交给workerThread去处理，它可以说是整个fork/join的容器。
ForkJoinWorkerThread：fork/join里面真正干活的工人，本质是一个线程。里面有一个ForkJoinPool.workQueue的队列存放这它要干的活，在开始接活之前它要向ForkJoinPool注册（registerWorker），拿到相应的workQueue。然后就从workQueue里面拿任务来处理。他是依附于ForkJoinPool而存活，如果ForkJoinPool销毁了，它也会跟着结束。
ForkJoinPool.workQueue：双端队列就是它，它负责存储接收的任务。
ForkJoinTask：代表fork/join里面的任务类型，我们一般用它的两个子类RecursiveTask、RecursiveAction。这两个区别在于RecursiveTask任务有返回值，RecursiveAction没有返回值。任务的处理逻辑包括任务的切分都集中在compute()方法里面。

工作方式

使用一直分治算法，递归地将任务分割成更小的子任务，其中阈值可配置，然后把子任务分配给不同的线程并发执行，最后再把结果组合起来。该方法常见于数组于集合的运算。
由于提交的任务不一定能够递归的分割成ForkJoinTask，且ForkJoinTask执行时间不等长，所以ForkJoinPool使用一种工作窃取的算法，允许空闲的线程“窃取”分配给另一线程的工作，由于工作无法平均分配并执行，所以工作窃取算法能更高效地利用硬件资源。

流程细节

Fork/Join框架与线程池的区别

ForkJoinPool不是为了替代ExecutorService，而是它的补充，在某些场景下性能比ExecutorService更好。
ForkJoinPool主要用于实现“分而治之”的算法，特别是分治之后递归调用的函数，例如quick sort等。
ForkJoinPool最适合的是计算密集型的任务，如果存在I/O，线程同步，sleep()等会造成线程长时间阻塞的情况，最好配合使用ManagedBlocker。
采用“工作窃取”模式（work-stealing）
当执行新的任务时它可以将其拆分成更小的任务执行，并将小任务加到线程队列中，然后再从一个随机线程的队列中偷一个并把它放在自己的队列中，本质上是为了保证线程不闲着，保持所有cpu繁忙。
与一般线程池处理任务的方式不同。
相对于一般的线程池实现，fork/join框架的优势体现在对其中包含的任务的处理方式上。在一般的线程池中，如果一个线程正在执行的任务由于某些原因无法继续运行，那么该线程就会处于等待状态（每个线程处理任务是同步的）。而在fork/join框架实现中，如果某个子问题由于等待另外一个子问题的完成而无法继续运行，那么处理该子问题的线程（等待其实就是空闲）会主动寻找尚未运行的子问题来执行（空闲的线程会从一个双端队列的尾部窃取一个任务来执行，不会让自己闲着）。这种方式减少了线程的等待时间，提高了性能。

JDK8 对 Fork/Join 的优化

JDK8 对 Fork/Join 的优化：主要是让 Fork/Join 使用起来更加方便。对 Fork/Join 进行了封装，简化使用方式。

应用场景

ThreadPool：多见于线程并发，阻塞时延比较长的，这种线程池比较常用，一般设置的线程个数根据业务性能要求会比较多。
ForkJoinPool：特点是少量线程完成大量任务，一般用于非阻塞的，能快速处理的业务，或阻塞时延比较低的。

使用示例

package com.study.practice;import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
import java.util.stream.LongStream;/*** @Description : test* @Version : V1.0.0* @Date : 2022/4/3 11:35*/
public class Test {public static void main(String[] args) {// jdk7的写法long startTime = System.currentTimeMillis();ForkJoinPool forkJoinPool = new ForkJoinPool();ForkJoinCalculate forkJoinCalculate = new ForkJoinCalculate(0, 1000000000L);Long res = forkJoinPool.invoke(forkJoinCalculate);System.out.println("jdk7 result " + res + " cost " + (System.currentTimeMillis() - startTime));// jdk8的写法long start = System.currentTimeMillis();// 只有并行流才会使用fork/join框架，否则就是单线程执行sum = LongStream.rangeClosed(0, 1000000000L).parallel().sum();System.out.println("jdk8 parallel exe result " + sum + " cost " + (System.currentTimeMillis() - start));}private static class ForkJoinCalculate extends RecursiveTask<Long> {private long start;private long end;private static final long THRESHOLD = 10000;public ForkJoinCalculate(long start, long end) {this.start = start;this.end = end;}@Overrideprotected Long compute() {if (end - start <= THRESHOLD) {long sum = 0;for (long i = start; i <= end; i++) {sum += i;}return sum;} else {long middle = (start + end) / 2;ForkJoinCalculate leftFork = new ForkJoinCalculate(start, middle);// 把任务分配到异步线程池中leftFork.fork();ForkJoinCalculate rightFork = new ForkJoinCalculate(middle + 1, end);// 把任务分配到异步线程池中rightFork.fork();// 把结果合并return leftFork.join() + rightFork.join();}}}
}

执行结果：

jdk7 result 500000000500000000 cost 2438
jdk8 parallel exe result 500000000500000000 cost 93

参考

线程池之ThreadPool与ForkJoinPool
Java-ForkJoinPool详解
【小家java】Java线程池之—ForkJoinPool线程池的使用以及原理

线程池之ForkJoinPool相关推荐

线程及同步的性能 – 线程池/ ThreadPoolExecutors/ ForkJoinPool
线程池和ThreadPoolExecutors 虽然在程序中可以直接使用Thread类型来进行线程操作,但是更多的情况是使用线程池,尤其是在Java EE应用服务器中,一般会使用若干个线程池来处理来自 ...
java8 forkjoinpool_Java 普通线程池与 ForkJoinPool 的效果对比
Java 多线程编程常用的一个接口是 ExecutorService, 其实就一个线程池的接口,一般由两种方式创建线程池,一为 Executors 的工厂方法,二则创建 ForkJoinPool 实例 ...
Tomcat线程池和ForkJoinPool线程池
目录 Tomcat线程池这里线程分了三个部分: Tomcat线程池与ThreadPoolExecutor的区别: 一些注意参数: Fork/Join线程池 ForkJoinPool 发送信息例子: ...
ForkJoinPool线程池
在Java 7中引入了一种新的线程池:ForkJoinPool. 它同ThreadPoolExecutor一样,也实现了Executor和ExecutorService接口.它使用了一个无限队列来保存 ...
c++ 线程池_基础篇：高并发一瞥，线程和线程池的总结
进程是执行程序的实体,拥有独属的进程空间(内存.磁盘等).而线程是进程的一个执行流程,一个进程可包含多个线程,共享该进程的所有资源:代码段,数据段(全局变量和静态变量),堆存储:但每个线程拥有自己的执 ...
线程池参数详解_java中常见的六种线程池详解
之前我们介绍了线程池的四种拒绝策略,了解了线程池参数的含义,那么今天我们来聊聊Java 中常见的几种线程池,以及在jdk7 加入的 ForkJoin 新型线程池首先我们列出Java 中的六种线程池如 ...
血的教训--如何正确使用线程池submit和execute方法
血的教训之背景:使用线程池对存量数据进行迁移,但是总有一批数据迁移失败,无异常日志打印凶案起因听说parallelStream并行流是个好东西,由于日常开发stream串行流的场景比较多,这次 ...
Java线程池原理与实例详解
Wiki 采用new Thread的方式产生多线程,可能有以下一些问题:  线程的创建和销毁开销很大,尤其是有些线程的存在时间较短:  线程的创建和销毁过程中伴随着CPU在线程间的切换,开销很大: ...
java 异步线程池_Java - 异步线程池
一.异步线程启动: new Thread newThread(newRunnable() { @Overridepublic voidrun() {//-- 这里是异步线程内的逻辑 } } ).sta ...

线程池之ForkJoinPool

概述

ThreadPool与ForkJoinPool介绍

ThreadPool Executor

工作方式

流程细节

ForkJoinPool Executor

ForkJoinPool组成类

工作方式

流程细节

Fork/Join框架与线程池的区别

JDK8 对 Fork/Join 的优化

应用场景

使用示例

参考

线程池之ForkJoinPool相关推荐

最新文章

热门文章