[Java 并发]你知道不知道 Fork/Join ?

Dear 丶 2023-02-20 07:29 9阅读 0赞

如果要说 Fork/Join 的话,就得来说说分治.  
分治分治,就是分而治之嘛,具体一点儿就是把一个复杂的问题分解成多个相似的子问题,然后呢,再把子问题分解成更小的子问题,直到子问题简单到可以直接求解才算结束  
这种思想,是不是让你想起了归并排序/快速排序/二分查找?没错,这些算法的实现也是借助了分治的思想

分治分治,估计可以猜出来吧,最重要的就是两点:一个是分,一个是治  
分什么呢?就是把一个复杂的问题分解成子问题,直到子问题可以直接求解结束  
治什么呢?刚刚把分解的子问题求解出来了对吧?那是不是要汇总一下,最后求出总问题的解?  
千言万语不如一张图,那就来张图:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3psbF8wNDA1_size_16_color_FFFFFF_t_70_pic_center]

看完有没有觉得,哦,原来这就是 fork/join 的赶脚?  
如果有的话,那我这张图就没白画,头发没白掉~

# ForkJoinTask #

Fork/Join 是一个并行计算的框架,主要就是用来支持分治任务模型的  
Fork/Join 计算框架主要包含两部分,一部分是**分治任务的线程池 ForkJoinPool**  ,另一部分是**分治任务 ForkJoinTask**  ,先来看 ForkJoinTask  
因为如果你想要 join 的话,是不是要先有 fork ?而 fork 方法在 ForkJoinTask 里面,所以咱们先来看看它

在源码中,能够看到 fork 方法:

public final ForkJoinTask<V> fork() { 
            Thread t;
            // ForkJoinWorkerThread 是执行 ForkJoinTask 的专有线程
            // 判断当前线程是否是 ForkJoin 专有线程,如果是则将任务 push 到当前线程所负责的队列里面去
            if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
                ((ForkJoinWorkerThread)t).workQueue.push(this);
            else
            	// 如果不是 ForkJoin 专有线程,就将任务提交到默认的 common 线程池中
                ForkJoinPool.common.externalPush(this);
            return this;
        }

仔细看的话, fork() 就做了一件事,就是把**任务放到当前工作线程的工作队列中**  
咱们继续往下看, fork() 结束,是怎么做的 join()

public final V join() { 
            int s;
            // 调用 dojoin 方法来获取当前任务的执行状态
            if ((s = doJoin() & DONE_MASK) != NORMAL)
            	// 任务异常,抛出异常
                reportException(s);
            // 任务完成,返回结果
            return getRawResult();
        }

能够看到, join() 方法就是等待处理任务的线程结束,然后拿到返回值  
在 join 方法中调用了 doJoin 方法,咱们来瞅瞅它具体是个啥

private int doJoin() { 
            int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w;
            // 首先判断任务是否执行完毕,如果执行完毕,直接返回结果就可以了
            return (s = status) < 0 ? s :
            	// 如果没有执行完毕,接下来要判断是不是 ForkJoinWorkerThread 线程
                ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
                // 如果是,判断这个任务是不是在工作队列最前边(也就是下一个执行的就是它)
                // tryUnpush() 方法判断任务是不是在工作队列最前边,是的话就返回 true
                // doExec() 方法执行任务
                (w = (wt = (ForkJoinWorkerThread)t).workQueue).
                // 如果是在工作队列最前边,并且任务执行完毕,直接返回结果即可
                tryUnpush(this) && (s = doExec()) < 0 ? s :
                // 如果不是工作队列最前边或者任务没有执行完毕,调用 awaitJoin() 来执行任务
                // awaitJoin(): 使用自旋使得任务执行完成,返回结果
                wt.pool.awaitJoin(w, this, 0L) :
                // 如果不是 ForkJoinWorkThread 线程,执行 externalAwaitDone() 返回任务结果
                externalAwaitDone();
        }

我知道,你看完上面的分析之后,就懵逼了  
因为我当时分析这块还理解了好久  
那如果来张图呢?  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3psbF8wNDA1_size_16_color_FFFFFF_t_70_pic_center 1]

ForkJoinTask 有两个子类 --> RecursiveAction & RecursiveTask ,它们都是通过递归的方式来处理分治任务的,这两个子类都定义了抽象方法 compute() ,不过区别就是 RecursiveAction 定义的 compute() 没有返回值,而 RecursiveTask 定义的 compute() 方法是有返回值的

# ForkJoinPool #

ForkJoinPool 是用来执行 ForkJoinTask 任务的线程池,它负责管理线程池中的线程和任务队列,还有就是线程池是否还接受任务,显示线程的运行状态也是在这里处理  
ForkJoinPool 本质上是一个 生产者-消费者 的实现,但是它更加的只能,因为它可以窃取别的任务,也就是说,如果一个工作线程空闲了,那么它可以"窃取"其他工作队列中的任务来做  
接下来咱们瞅瞅 ForkJoinPool 的源码

/** * Creates a {@code ForkJoinPool} with the given parameters, without * any security checks or parameter validation. Invoked directly by * makeCommonPool. */
         // 私有构造方法,没有任何的安全检查和参数校验,由 makeCommonPool 直接调用
         // parallelism 并行度
        private ForkJoinPool(int parallelism,
                             ForkJoinWorkerThreadFactory factory,
                             UncaughtExceptionHandler handler,
                             int mode,
                             String workerNamePrefix) { 
            this.workerNamePrefix = workerNamePrefix;
            this.factory = factory;
            this.ueh = handler;
            this.config = (parallelism & SMASK) | mode;
            long np = (long)(-parallelism); // offset ctl counts
            this.ctl = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK);
        }

ForkJoinPool 由 makeCommonPool 直接调用,来瞅瞅:

/** * Creates and returns the common pool, respecting user settings * specified via system properties. */
        private static ForkJoinPool makeCommonPool() { 
            int parallelism = -1;
            ForkJoinWorkerThreadFactory factory = null;
            UncaughtExceptionHandler handler = null;
            // 通过系统指定相关参数
            try {   // ignore exceptions in accessing/parsing properties
                String pp = System.getProperty
                    ("java.util.concurrent.ForkJoinPool.common.parallelism");
                String fp = System.getProperty
                    ("java.util.concurrent.ForkJoinPool.common.threadFactory");
                String hp = System.getProperty
                    ("java.util.concurrent.ForkJoinPool.common.exceptionHandler");
                if (pp != null)
                    parallelism = Integer.parseInt(pp);
                if (fp != null)
                    factory = ((ForkJoinWorkerThreadFactory)ClassLoader.
                               getSystemClassLoader().loadClass(fp).newInstance());
                if (hp != null)
                    handler = ((UncaughtExceptionHandler)ClassLoader.
                               getSystemClassLoader().loadClass(hp).newInstance());
            } catch (Exception ignore) { 
            }
            // 如果 factory 为空
            if (factory == null) { 
            	// 如果 getSecurityManager 返回 null ,说明系统还没有为当前应用程序建立安全管理器
            	// 使用 defaultForkJoinWorkerThreadFactory 来进行创建
                if (System.getSecurityManager() == null)
                    factory = defaultForkJoinWorkerThreadFactory;
                else // use security-managed default
                	// 如果 getSecurityManager 返回不是 null ,说明系统为当前应用程序建立好了安全管理器
                	// 使用 InnocuousForkJoinWorkerThreadFactory 来进行创建
                    factory = new InnocuousForkJoinWorkerThreadFactory();
            }
            if (parallelism < 0 && // default 1 less than #cores
                (parallelism = Runtime.getRuntime().availableProcessors() - 1) <= 0)
                parallelism = 1;
            if (parallelism > MAX_CAP)
                parallelism = MAX_CAP;
            return new ForkJoinPool(parallelism, factory, handler, LIFO_QUEUE,
                                    "ForkJoinPool.commonPool-worker-");
        }

这篇文章到这里,我就要划一个句号了  
摸着良心说,这篇文章内容,相对来说不是很多,因为很多源码都还没分析到,比如使用的 `workQueues` 队列,比如如何实现的任务窃取,都还没说到,等我回头再写一篇文章出来  
主要是分析源码,太费脑细胞 & 头发了 ٩(º﹃º٩)  
再加上最近有点儿放飞自我,沉迷于动漫中,等我刷完了动漫我再把细节补充上来

就酱~  
感谢您的阅读哇

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3psbF8wNDA1_size_16_color_FFFFFF_t_70_pic_center]: https://img-blog.csdnimg.cn/20200626121302158.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3psbF8wNDA1,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3psbF8wNDA1_size_16_color_FFFFFF_t_70_pic_center 1]: https://img-blog.csdnimg.cn/20200626161427428.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3psbF8wNDA1,size_16,color_FFFFFF,t_70#pic_center