ForkJoin框架源码分析（详细）

ForkJoin简介及使用

ForkJoin框架是CompletableFuture和java8 stream使用到的框架。主要用于分片处理的场景。
可以通过自定义分片粒度来实现任务分解。并行处理数据。

CompletableFuture能够实现响应式编程。但未用到ForkJoin的分片。所以对于CompletableFuture来说，ForkJoin仅是一个公用的线程池而已

stream能让java处理数据更加优雅，并行流stream使用了ForkJoin的分片处理(参考java.util.stream.ForEachTask等类)

ForkJoin源码分析

从CompletableFuture的使用进入

CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {System.out.println("hello world");return 0;
});CompletableFuture<Void> completableFuture1 = CompletableFuture.runAsync(() ->System.out.println("hello world")
);completableFuture1 = CompletableFuture.runAsync(() ->System.out.println("hello world"), Executors.newSingleThreadExecutor());completableFuture.get();

CompletableFuture提供了提交Runnable和Callable任务的接口。分别对应runAsync(Runnable r)和supplyAsync(Callable c)。runAsync(Runnable r)有一个重载方法runAsync(Runnable r, Executor e) 支持自定义线程池。如果调用runAsync(Runnable r)方法，则代表使用ForkJoinPool作为线程池。

ForkJoinPool源码剖析

public static CompletableFuture<Void> runAsync(Runnable runnable) {return asyncRunStage(asyncPool, runnable);
}

asyncPool参数解析

// ForkJoinPool.parallelism默认为可用CPU的数量
// parallelism可以自行设置
private static final boolean useCommonPool =(ForkJoinPool.getCommonPoolParallelism() > 1);// 当parallelism并行度等于1(不并行处理)时使用ThreadPerTaskExecutor线程池
private static final Executor asyncPool = useCommonPool ?ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();// ForkJoinPool每个任务可能有多个线程进行处理
// ThreadPerTaskExecutor每个任务创建一个线程来进行处理
static final class ThreadPerTaskExecutor implements Executor {public void execute(Runnable r) { new Thread(r).start(); }
}

ForkJoinPool的类图

CompletableFuture#asyncRunStage方法内容

static CompletableFuture<Void> asyncRunStage(Executor e, Runnable f) {if (f == null) throw new NullPointerException();CompletableFuture<Void> d = new CompletableFuture<Void>();// 这里调用的是ForkJoinPool#execute(Runnable task)方法e.execute(new AsyncRun(d, f));return d;
}

AsyncRun继承了ForkJoinTask类，实现了Runnable接口

static final class AsyncRun extends ForkJoinTask<Void>implements Runnable, AsynchronousCompletionTask

ForkJoinPool#execute(Runnable task)方法

public void execute(Runnable task) {if (task == null)throw new NullPointerException();ForkJoinTask<?> job;if (task instanceof ForkJoinTask<?>) // avoid re-wrap// 从CompletableFuture提交的任务会走到这job = (ForkJoinTask<?>) task;elsejob = new ForkJoinTask.RunnableExecuteAction(task);// 将任务放入工作队列externalPush(job);
}

ForkJoinPool#externalPush(ForkJoinTask<?> task)方法，将任务放入工作队列

final void externalPush(ForkJoinTask<?> task) {WorkQueue[] ws; WorkQueue q; int m;int r = ThreadLocalRandom.getProbe();int rs = runState;// 如果工作队列不为空(ws = workQueues) != null && (m = (ws.length - 1)) >= 0// q = ws[m & r & SQMASK]，q为ws中随机的偶数索引。// SQMASK为0x007e 126二进制位为11111110所以q的结果永远为偶数// U.compareAndSwapInt(q, QLOCK, 0, 1)加锁，QLOCK为第q个WorkQueue对象的锁的引用if ((ws = workQueues) != null && (m = (ws.length - 1)) >= 0 &&(q = ws[m & r & SQMASK]) != null && r != 0 && rs > 0 &&U.compareAndSwapInt(q, QLOCK, 0, 1)) {ForkJoinTask<?>[] a; int am, n, s;if ((a = q.array) != null &&(am = a.length - 1) > (n = (s = q.top) - q.base)) {int j = ((am & s) << ASHIFT) + ABASE;// 把任务放入WorkQueue[q].array的队尾U.putOrderedObject(a, j, task);// 队列的top++；U.putOrderedInt(q, QTOP, s + 1);// 解锁U.putIntVolatile(q, QLOCK, 0);// 处于active的工作线程不够则新建work来if (n <= 1)signalWork(ws, q);return;}// 解锁U.compareAndSwapInt(q, QLOCK, 1, 0);}// 如果工作队列为空，或者加锁失败，则初始化工作队列后提交任务externalSubmit(task);
}

private void externalSubmit(ForkJoinTask<?> task) {int r;                                    // initialize caller's probeif ((r = ThreadLocalRandom.getProbe()) == 0) {ThreadLocalRandom.localInit();r = ThreadLocalRandom.getProbe();}// 循环退出条件// 抛出RejectedExecutionException异常(runState为关闭状态)// 任务提交成功for (;;) {WorkQueue[] ws; WorkQueue q; int rs, m, k;boolean move = false;if ((rs = runState) < 0) {tryTerminate(false, false);     // help terminatethrow new RejectedExecutionException();}// 没初始化，就初始化WorkQueueelse if ((rs & STARTED) == 0 ||     // initialize((ws = workQueues) == null || (m = ws.length - 1) < 0)) {int ns = 0;// 加锁，锁标志位低位置为1， rs |= 1;rs = lockRunState();try {// 没初始化，则新建一个WorkQueueif ((rs & STARTED) == 0) {U.compareAndSwapObject(this, STEALCOUNTER, null,new AtomicLong());// 创建一个size为2的n次方的WorkQueue数组int p = config & SMASK; // ensure at least 2 slotsint n = (p > 1) ? p - 1 : 1;n |= n >>> 1; n |= n >>> 2;  n |= n >>> 4;n |= n >>> 8; n |= n >>> 16; n = (n + 1) << 1;workQueues = new WorkQueue[n];// 将置为已初始化状态ns = STARTED;}} finally {// 解锁并设置runState为已初始化状态unlockRunState(rs, (rs & ~RSLOCK) | ns);}}// 判断该任务所在WorkQueue是否为空else if ((q = ws[k = r & m & SQMASK]) != null) {if (q.qlock == 0 && U.compareAndSwapInt(q, QLOCK, 0, 1)) {ForkJoinTask<?>[] a = q.array;int s = q.top;boolean submitted = false; // initial submission or resizingtry {                      // locked version of pushif ((a != null && a.length > s + 1 - q.base) ||(a = q.growArray()) != null) {int j = (((a.length - 1) & s) << ASHIFT) + ABASE;U.putOrderedObject(a, j, task);U.putOrderedInt(q, QTOP, s + 1);submitted = true;}} finally {U.compareAndSwapInt(q, QLOCK, 1, 0);}// 任务提交成功if (submitted) {// 如果worker太少，则创建worker，没有空闲的worker则啥也不干。signalWork(ws, q);return;}}// 未提交成功则继续此循环move = true;                   // move on failure}else if (((rs = runState) & RSLOCK) == 0) { // create new queueq = new WorkQueue(this, null);q.hint = r;q.config = k | SHARED_QUEUE;q.scanState = INACTIVE;rs = lockRunState();           // publish indexif (rs > 0 &&  (ws = workQueues) != null &&k < ws.length && ws[k] == null)ws[k] = q;                 // else terminatedunlockRunState(rs, rs & ~RSLOCK);}elsemove = true;                   // move if busyif (move)r = ThreadLocalRandom.advanceProbe(r);}
}