Fork/Join framework and calling method in Java

What is ForkJoin?

ForkJoin Literally, Fork means bifurcation, and Join means combination. We can understand it as combining large tasks Split into small tasks for calculation and solution, and finally combine the results of the small tasks to find the solution to the large task. We can hand over these fissioned small tasks to different threads for calculation, which is the principle of distributed computing. A thought. This is similar to distributed offline computing MapReduce in big data. One of the most classic applications of ForkJoin is Stream in Java8. We know that Stream is divided into serial stream and parallel stream. The parallel stream parallelStream relies on ForkJoin to achieve parallel processing. of.

Let’s take a look at the core ForkJoinTask and ForkJoinPool.

ForkJoinTask task

The dependencies of ForkJoinTask itself are not complicated. It implements the Future interface just like the asynchronous task calculation FutureTask

Let's study the core source code of ForkJoinTask and how the task is calculated through the divide and conquer method.

The core of ForkJoinTask is the fork() and join() methods.

fork()

• Determine whether the current thread is a ForkJoinWorkerThread thread

• Yes Directly push the current thread into the work queue

• Whether to call the externalPush method of ForkJoinPool

##In

ForkJoinPool A static common object is constructed, and what is called here is commonexternalPush()

#join()

• Call the doJoin() method and wait for the thread execution to complete

    public final ForkJoinTask<V> fork() {
        Thread t;
        if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
            ((ForkJoinWorkerThread)t).workQueue.push(this);
        else
            ForkJoinPool.common.externalPush(this);
        return this;
    }

    public final V join() {
        int s;
        if ((s = doJoin() & DONE_MASK) != NORMAL)
            reportException(s);
        return getRawResult();
    }

    private int doJoin() {
        int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w;
        return (s = status) < 0 ? s :
            ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
            (w = (wt = (ForkJoinWorkerThread)t).workQueue).
            tryUnpush(this) && (s = doExec()) < 0 ? s :
            wt.pool.awaitJoin(w, this, 0L) :
            externalAwaitDone();
    }

	// 获取结果的方法由子类实现
	public abstract V getRawResult();	

RecursiveTask is a subclass of ForkJoinTask that mainly implements the method of obtaining results through generic constraints result. If we need to create a task ourselves, we still need to implement RecursiveTask and write the core calculation method compute().

public abstract class RecursiveTask<V> extends ForkJoinTask<V> {
    private static final long serialVersionUID = 5232453952276485270L;

    V result;

    protected abstract V compute();

    public final V getRawResult() {
        return result;
    }

    protected final void setRawResult(V value) {
        result = value;
    }
    protected final boolean exec() {
        result = compute();
        return true;
    }

}

ForkJoinPool thread pool

Many functions in ForkJoinTask rely on the ForkJoinPool thread pool, so ForkJoinTask cannot run without ForkJoinPool. ForkJoinPool has many similarities with ThreadPoolExecutor. It is specially used for The thread pool that executes the ForkJoinTask task. I have previously introduced the thread pool technology in an article. If you are interested, you can read it - analyze the implementation principle of the thread pool (pooling technology) from the java source code

ForkJoinPool The inheritance relationship with ThreadPoolExecutor is almost the same, they are equivalent to brothers.

Work stealing algorithm

The work stealing algorithm is adopted in ForkJoinPool. If a new thread is created for each fork subtask, the system resources will be affected. The overhead is huge, so a thread pool must be used. A general thread pool has only one task queue, but for ForkJoinPool, since the subtasks forked out by the same task are parallel, in order to improve efficiency and reduce thread competition, these parallel tasks need to be placed in different queues. , since threads process different tasks at different speeds, there may be a thread that has finished executing the tasks in its own queue first. At this time, in order to improve efficiency, the thread can be allowed to "

steal" other task queues This is the so-called "work-stealing algorithm".

For general queues, the elements entering the queue are at the end of the queue, and the elements leaving the queue are at the beginning. To meet the needs of "work stealing", the task queue should support dequeuing elements from the "tail" , which can reduce conflicts with other worker threads (because other worker threads will get the tasks in their own task queues from the head of the team). In this case, a double-ended blocking queue needs to be used to solve the problem.

Construction method

First let’s look at the construction method of the ForkJoinPool thread pool. It provides us with three forms of construction, the most complex of which is the construction of four input parameters. Next we Take a look at what its four input parameters represent?

• int parallelism parallelism level (does not represent the maximum number of existing threads)

• ForkJoinWorkerThreadFactory factory Thread creation factory

• UncaughtExceptionHandler handler Exception capture handler

• boolean asyncMode first-in-first-out working mode or last-in-first-out working mode

    public ForkJoinPool() {
        this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()),
             defaultForkJoinWorkerThreadFactory, null, false);
    }

	public ForkJoinPool(int parallelism) {
        this(parallelism, defaultForkJoinWorkerThreadFactory, null, false);
    }

	public ForkJoinPool(int parallelism,
                        ForkJoinWorkerThreadFactory factory,
                        UncaughtExceptionHandler handler,
                        boolean asyncMode) {
        this(checkParallelism(parallelism),
             checkFactory(factory),
             handler,
             asyncMode ? FIFO_QUEUE : LIFO_QUEUE,
             "ForkJoinPool-" + nextPoolId() + "-worker-");
        checkPermission();
    }

Submission method

Let’s take a look at the method of submitting a task:

externalPushThis method is very familiar to us, it is exactly when forking If the current thread is not ForkJoinWorkerThread, the newly submitted task will also be executed through this method. It can be seen that fork is to create a new subtask for submission.

externalSubmit是最为核心的一个方法，它可以首次向池提交第一个任务，并执行二次初始化。它还可以检测外部线程的首次提交，并创建一个新的共享队列。

signalWork(ws, q)是发送工作信号，让工作队列进行运转。

    public ForkJoinTask<?> submit(Runnable task) {
        if (task == null)
            throw new NullPointerException();
        ForkJoinTask<?> job;
        if (task instanceof ForkJoinTask<?>) // avoid re-wrap
            job = (ForkJoinTask<?>) task;
        else
            job = new ForkJoinTask.AdaptedRunnableAction(task);
        externalPush(job);
        return job;
    }

    final void externalPush(ForkJoinTask<?> task) {
        WorkQueue[] ws; WorkQueue q; int m;
        int r = ThreadLocalRandom.getProbe();
        int rs = runState;
        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;
                U.putOrderedObject(a, j, task);
                U.putOrderedInt(q, QTOP, s + 1);
                U.putOrderedInt(q, QLOCK, 0);
                if (n <= 1)
                    signalWork(ws, q);
                return;
            }
            U.compareAndSwapInt(q, QLOCK, 1, 0);
        }
        externalSubmit(task);
    }

    private void externalSubmit(ForkJoinTask<?> task) {
        int r;                                    // initialize caller&#39;s probe
        if ((r = ThreadLocalRandom.getProbe()) == 0) {
            ThreadLocalRandom.localInit();
            r = ThreadLocalRandom.getProbe();
        }
        for (;;) {
            WorkQueue[] ws; WorkQueue q; int rs, m, k;
            boolean move = false;
            if ((rs = runState) < 0) {
                tryTerminate(false, false);     // help terminate
                throw new RejectedExecutionException();
            }
            else if ((rs & STARTED) == 0 ||     // initialize
                     ((ws = workQueues) == null || (m = ws.length - 1) < 0)) {
                int ns = 0;
                rs = lockRunState();
                try {
                    if ((rs & STARTED) == 0) {
                        U.compareAndSwapObject(this, STEALCOUNTER, null,
                                               new AtomicLong());
                        // create workQueues array with size a power of two
                        int p = config & SMASK; // ensure at least 2 slots
                        int 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 {
                    unlockRunState(rs, (rs & ~RSLOCK) | ns);
                }
            }
            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 resizing
                    try {                      // locked version of push
                        if ((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) {
                        signalWork(ws, q);
                        return;
                    }
                }
                move = true;                   // move on failure
            }
            else if (((rs = runState) & RSLOCK) == 0) { // create new queue
                q = new WorkQueue(this, null);
                q.hint = r;
                q.config = k | SHARED_QUEUE;
                q.scanState = INACTIVE;
                rs = lockRunState();           // publish index
                if (rs > 0 &&  (ws = workQueues) != null &&
                    k < ws.length && ws[k] == null)
                    ws[k] = q;                 // else terminated
                unlockRunState(rs, rs & ~RSLOCK);
            }
            else
                move = true;                   // move if busy
            if (move)
                r = ThreadLocalRandom.advanceProbe(r);
        }
    }

创建工人（线程）

提交任务后，通过signalWork(ws, q)方法，发送工作信号，当符合没有执行完毕，且没有出现异常的条件下，循环执行任务，根据控制变量尝试添加工人（线程），通过线程工厂，生成线程，并且启动线程，也控制着工人（线程）的下岗。

    final void signalWork(WorkQueue[] ws, WorkQueue q) {
        long c; int sp, i; WorkQueue v; Thread p;
        while ((c = ctl) < 0L) {                       // too few active
            if ((sp = (int)c) == 0) {                  // no idle workers
                if ((c & ADD_WORKER) != 0L)            // too few workers
                    tryAddWorker(c);
                break;
            }
            if (ws == null)                            // unstarted/terminated
                break;
            if (ws.length <= (i = sp & SMASK))         // terminated
                break;
            if ((v = ws[i]) == null)                   // terminating
                break;
            int vs = (sp + SS_SEQ) & ~INACTIVE;        // next scanState
            int d = sp - v.scanState;                  // screen CAS
            long nc = (UC_MASK & (c + AC_UNIT)) | (SP_MASK & v.stackPred);
            if (d == 0 && U.compareAndSwapLong(this, CTL, c, nc)) {
                v.scanState = vs;                      // activate v
                if ((p = v.parker) != null)
                    U.unpark(p);
                break;
            }
            if (q != null && q.base == q.top)          // no more work
                break;
        }
    }

    private void tryAddWorker(long c) {
        boolean add = false;
        do {
            long nc = ((AC_MASK & (c + AC_UNIT)) |
                       (TC_MASK & (c + TC_UNIT)));
            if (ctl == c) {
                int rs, stop;                 // check if terminating
                if ((stop = (rs = lockRunState()) & STOP) == 0)
                    add = U.compareAndSwapLong(this, CTL, c, nc);
                unlockRunState(rs, rs & ~RSLOCK);
                if (stop != 0)
                    break;
                if (add) {
                    createWorker();
                    break;
                }
            }
        } while (((c = ctl) & ADD_WORKER) != 0L && (int)c == 0);
    }

    private boolean createWorker() {
        ForkJoinWorkerThreadFactory fac = factory;
        Throwable ex = null;
        ForkJoinWorkerThread wt = null;
        try {
            if (fac != null && (wt = fac.newThread(this)) != null) {
                wt.start();
                return true;
            }
        } catch (Throwable rex) {
            ex = rex;
        }
        deregisterWorker(wt, ex);
        return false;
    }

   final void deregisterWorker(ForkJoinWorkerThread wt, Throwable ex) {
        WorkQueue w = null;
        if (wt != null && (w = wt.workQueue) != null) {
            WorkQueue[] ws;                           // remove index from array
            int idx = w.config & SMASK;
            int rs = lockRunState();
            if ((ws = workQueues) != null && ws.length > idx && ws[idx] == w)
                ws[idx] = null;
            unlockRunState(rs, rs & ~RSLOCK);
        }
        long c;                                       // decrement counts
        do {} while (!U.compareAndSwapLong
                     (this, CTL, c = ctl, ((AC_MASK & (c - AC_UNIT)) |
                                           (TC_MASK & (c - TC_UNIT)) |
                                           (SP_MASK & c))));
        if (w != null) {
            w.qlock = -1;                             // ensure set
            w.transferStealCount(this);
            w.cancelAll();                            // cancel remaining tasks
        }
        for (;;) {                                    // possibly replace
            WorkQueue[] ws; int m, sp;
            if (tryTerminate(false, false) || w == null || w.array == null ||
                (runState & STOP) != 0 || (ws = workQueues) == null ||
                (m = ws.length - 1) < 0)              // already terminating
                break;
            if ((sp = (int)(c = ctl)) != 0) {         // wake up replacement
                if (tryRelease(c, ws[sp & m], AC_UNIT))
                    break;
            }
            else if (ex != null && (c & ADD_WORKER) != 0L) {
                tryAddWorker(c);                      // create replacement
                break;
            }
            else                                      // don&#39;t need replacement
                break;
        }
        if (ex == null)                               // help clean on way out
            ForkJoinTask.helpExpungeStaleExceptions();
        else                                          // rethrow
            ForkJoinTask.rethrow(ex);
    }

    public static interface ForkJoinWorkerThreadFactory {
        public ForkJoinWorkerThread newThread(ForkJoinPool pool);
    }
    static final class DefaultForkJoinWorkerThreadFactory
        implements ForkJoinWorkerThreadFactory {
        public final ForkJoinWorkerThread newThread(ForkJoinPool pool) {
            return new ForkJoinWorkerThread(pool);
        }
    }
    protected ForkJoinWorkerThread(ForkJoinPool pool) {
        // Use a placeholder until a useful name can be set in registerWorker
        super("aForkJoinWorkerThread");
        this.pool = pool;
        this.workQueue = pool.registerWorker(this);
    }

    final WorkQueue registerWorker(ForkJoinWorkerThread wt) {
        UncaughtExceptionHandler handler;
        wt.setDaemon(true);                           // configure thread
        if ((handler = ueh) != null)
            wt.setUncaughtExceptionHandler(handler);
        WorkQueue w = new WorkQueue(this, wt);
        int i = 0;                                    // assign a pool index
        int mode = config & MODE_MASK;
        int rs = lockRunState();
        try {
            WorkQueue[] ws; int n;                    // skip if no array
            if ((ws = workQueues) != null && (n = ws.length) > 0) {
                int s = indexSeed += SEED_INCREMENT;  // unlikely to collide
                int m = n - 1;
                i = ((s << 1) | 1) & m;               // odd-numbered indices
                if (ws[i] != null) {                  // collision
                    int probes = 0;                   // step by approx half n
                    int step = (n <= 4) ? 2 : ((n >>> 1) & EVENMASK) + 2;
                    while (ws[i = (i + step) & m] != null) {
                        if (++probes >= n) {
                            workQueues = ws = Arrays.copyOf(ws, n <<= 1);
                            m = n - 1;
                            probes = 0;
                        }
                    }
                }
                w.hint = s;                           // use as random seed
                w.config = i | mode;
                w.scanState = i;                      // publication fence
                ws[i] = w;
            }
        } finally {
            unlockRunState(rs, rs & ~RSLOCK);
        }
        wt.setName(workerNamePrefix.concat(Integer.toString(i >>> 1)));
        return w;
    }

例：ForkJoinTask实现归并排序

这里我们就用经典的归并排序为例，构建一个我们自己的ForkJoinTask，按照归并排序的思路，重写其核心的compute()方法，通过ForkJoinPool.submit(task)提交任务，通过get()同步获取任务执行结果。

package com.zhj.interview;

import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

public class Test16 {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        int[] bigArr = new int[10000000];
        for (int i = 0; i < 10000000; i++) {
            bigArr[i] = (int) (Math.random() * 10000000);
        }
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        MyForkJoinTask task = new MyForkJoinTask(bigArr);
        long start = System.currentTimeMillis();
        forkJoinPool.submit(task).get();
        long end = System.currentTimeMillis();
        System.out.println("耗时：" + (end-start));
	}

}
class MyForkJoinTask extends RecursiveTask<int[]> {

    private int source[];

    public MyForkJoinTask(int source[]) {
        if (source == null) {
            throw new RuntimeException("参数有误！！！");
        }
        this.source = source;
    }

    @Override
    protected int[] compute() {
        int l = source.length;
        if (l < 2) {
            return Arrays.copyOf(source, l);
        }
        if (l == 2) {
            if (source[0] > source[1]) {
                int[] tar = new int[2];
                tar[0] = source[1];
                tar[1] = source[0];
                return tar;
            } else {
                return Arrays.copyOf(source, l);
            }
        }
        if (l > 2) {
            int mid = l / 2;
            MyForkJoinTask task1 = new MyForkJoinTask(Arrays.copyOf(source, mid));
            task1.fork();
            MyForkJoinTask task2 = new MyForkJoinTask(Arrays.copyOfRange(source, mid, l));
            task2.fork();
            int[] res1 = task1.join();
            int[] res2 = task2.join();
            int tar[] = merge(res1, res2);
            return tar;
        }
        return null;
    }
	// 合并数组
    private int[] merge(int[] res1, int[] res2) {
        int l1 = res1.length;
        int l2 = res2.length;
        int l = l1 + l2;
        int tar[] = new int[l];
        for (int i = 0, i1 = 0, i2 = 0; i < l; i++) {
            int v1 = i1 >= l1 ? Integer.MAX_VALUE : res1[i1];
            int v2 = i2 >= l2 ? Integer.MAX_VALUE : res2[i2];
            // 如果条件成立，说明应该取数组array1中的值
            if(v1 < v2) {
                tar[i] = v1;
                i1++;
            } else {
                tar[i] = v2;
                i2++;
            }
        }
        return tar;
    }
}

ForkJoin计算流程

通过ForkJoinPool提交任务，获取结果流程如下，拆分子任务不一定是二分的形式，可参照MapReduce的模式，也可以按照具体需求进行灵活的设计。

The above is the detailed content of Fork/Join framework and calling method in Java. For more information, please follow other related articles on the PHP Chinese website!