Home > Java > javaTutorial > body text

Fork/Join framework and calling method in Java

WBOY
Release: 2023-05-08 11:10:07
forward
1633 people have browsed it

    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

    Fork/Join framework and calling method in Java

    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();	
      Copy after login
    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;
        }
    
    }
    Copy after login

    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.

    Fork/Join framework and calling method in Java

    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();
          }
      Copy after login
    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);
            }
        }
    Copy after login

    创建工人(线程)

    提交任务后,通过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;
        }
    Copy after login

    例: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;
        }
    }
    Copy after login

    ForkJoin计算流程

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

    Fork/Join framework and calling method in Java

    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!

    Related labels:
    source:yisu.com
    Statement of this Website
    The content of this article is voluntarily contributed by netizens, and the copyright belongs to the original author. This site does not assume corresponding legal responsibility. If you find any content suspected of plagiarism or infringement, please contact admin@php.cn
    Popular Tutorials
    More>
    Latest Downloads
    More>
    Web Effects
    Website Source Code
    Website Materials
    Front End Template