java forkjoin 简书_ForkJoinPool

分手后的思念是犯贱 2022-11-08 14:18 297阅读 0赞

为什么使用fork join框架？

ForkJoinPool是ExecutorSerice的一个补充，而不是替代品

并不适合所有场景；

特别适合用于“分而治之”，递归计算的算法；

JAVA8中CompeleteFuture、并发流等都是基于ForkJoinPool实现；

本文将讲述什么样的场景合适、什么样的场景不合适，并且深入剖析为啥某些场景不合适。

fork join框架使用简单实例

使用RecursiveTask实现一个累加的功能，使用分而治之的思想，实现分段求和后汇总

public class SumTask extends RecursiveTask \{

private Integer start = 0;

private Integer end = 0;

public SumTask(int start, int end) \{

this.start = start;

this.end = end;

@Override

protected Integer compute() \{

if (end - start < 100) \{

//小于100时直接返回结果

int sumResult = 0;

for (int i = start; i <= end; i++) \{

sumResult += i;

return sumResult;

\} else \{

//大于一百时进行分割

int middle = (end + start) / 2;

SumTask leftSum = new SumTask(this.start, middle);

SumTask rightSum = new SumTask(middle, this.end);

leftSum.fork();

rightSum.fork();

return leftSum.join() + rightSum.join();

public static void main(String\[\] args) \{

SumTask sumTask = new SumTask(1, 999999);

sumTask.fork();

System.out.print("result:" + sumTask.join());

默认线程池有固定的线程数,会根据可用的availableProcessors来计算线程数量；

/\*\*

\* Creates a \{@code ForkJoinPool\} with parallelism equal to \{@link

\* java.lang.Runtime\#availableProcessors\}, using the \{@linkplain

\* \#defaultForkJoinWorkerThreadFactory default thread factory\},

\* no UncaughtExceptionHandler, and non-async LIFO processing mode.

\* @throws SecurityException if a security manager exists and

\* the caller is not permitted to modify threads

\* because it does not hold \{@link

\* java.lang.RuntimePermission\}\{@code ("modifyThread")\}

\*/

public ForkJoinPool() \{

this(Math.min(MAX\_CAP, Runtime.getRuntime().availableProcessors()),

defaultForkJoinWorkerThreadFactory, null, false);

如果想使用自定义线程池，比如100个线程的线程池，可以如下：

当然无论你初始化多少线程，都只会有和CPU数量相等的几个线程运行，大部分场景下并无异于这个框架的加速；

public static void main(String\[\] args) \{

ForkJoinPool forkJoinPool=new ForkJoinPool(100);

SumTask sumTask = new SumTask(1, 999999);

forkJoinPool.submit(sumTask);

System.out.print("result:" + sumTask.join());

我们来看看fork的时候做了些什么

/\*\*

\* Arranges to asynchronously execute this task in the pool the

\* current task is running in, if applicable, or using the \{@link

\* ForkJoinPool\#commonPool()\} if not \{@link \#inForkJoinPool\}. While

\* it is not necessarily enforced, it is a usage error to fork a

\* task more than once unless it has completed and been

\* reinitialized. Subsequent modifications to the state of this

\* task or any data it operates on are not necessarily

\* consistently observable by any thread other than the one

\* executing it unless preceded by a call to \{@link \#join\} or

\* related methods, or a call to \{@link \#isDone\} returning \{@code

\* true\}.

\* @return \{@code this\}, to simplify usage

\*/

public final ForkJoinTask fork() \{

Thread t;

if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)

((ForkJoinWorkerThread)t).workQueue.push(this);

else

ForkJoinPool.common.externalPush(this);

return this;

当前线程不是个ForkJoinWorkerThread的时候，则加入到ForkJoinPool线程池(基于ExecutorService实现);

如果当前线程已经是个ForkJoinWorkerThread了，则把这个任务加入到当前线程的workQueue;

![a152c0a0d2d0][]

大概的图

这是和普通线程池不同的地方，task并不是交给线程池中的queue，而是放到线程本地的workQueue

我们来看看ForkJoinPool中的task是如何运行的

a. 线程以LIFO先进后出方式从本地队列获取任务，执行，直到自己的队列为空；

b. 查看其他ForkJoinWorkerThread是否有未执行task，有的话通过work−stealing窃取,窃取方式为FIFO先进先出，减少竞争；优先看曾今从自己那里窃取任务的thread，如果有的话；

c. 任务运行完成时，返回结果；

java8并发流

java8的并发流使用的也是ForkJoinPool

myList.parallelStream.map(obj -> longRunningOperation())

默认使用公用线程池，使用独立线程池的话如下

ForkJoinPool forkJoinPool = new ForkJoinPool(3);

forkJoinPool.submit(() -> \{

firstRange.parallelStream().forEach((number) -> \{

try \{

Thread.sleep(5);

\} catch (InterruptedException e) \{ \}

\});

ForkJoinPool forkJoinPool2 = new ForkJoinPool(3);

forkJoinPool2.submit(() -> \{

secondRange.parallelStream().forEach((number) -> \{

try \{

Thread.sleep(5);

\} catch (InterruptedException e) \{

\});

4.为什么不适合执行有block比如有io的任务

线程block的时候，线程池会调度线程池队列中的其他未线程运行，这是ExecutorService的机制；

但是ForkJoinPool中的ForkJoinWorkerThread工作机制是不停执行本地workQueue中的task，task是一个个取的，顺序执行，没有塞回去的动作，并不会因为某个task引起block后而换个task继续执行；

[a152c0a0d2d0]: /images/20221023/8f38112a194341818561a72fe925ce31.png

java forkjoin 简书_ForkJoinPool

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关 java forkjoin 简书_ForkJoinPool

相关 java选择排序简书_Java排序算法

相关 weex android 简书,Weex Extend

相关记简书--vuex学习

相关简书中添加代码框

相关 Java并发 - ForkJoinPool

相关高仿简书个人中心

相关 Spring Cloud【2】-简书

相关重拾简书

相关刷简书浏览量