ホームページ >Java >&#&チュートリアル >Java の Fork/Join フレームワークとメソッド呼び出し

Java の Fork/Join フレームワークとメソッド呼び出し

WBOY
WBOY転載
2023-05-08 11:10:071652ブラウズ

    ForkJoin とは?

    ForkJoin 文字通り、Fork は分岐を意味し、Join は結合を意味します。大きなタスクを組み合わせると理解できます。計算と解を繰り返し、最後に小さなタスクの結果を組み合わせて大きなタスクの解を求めるという、分裂した小さなタスクを別のスレッドに渡して計算することが分散コンピューティングの原理であると考えられます。これは、ビッグ データにおける分散オフライン コンピューティングの MapReduce に似ています。ForkJoin の最も古典的なアプリケーションの 1 つは、Java8 の Stream です。Stream はシリアル ストリームとパラレル ストリームに分割されることがわかっています。パラレル ストリームのParallelStream は、ForkJoin に依存して並列処理を実現します。の。

    コアの ForkJoinTaskForkJoinPool を見てみましょう。

    ForkJoinTask タスク

    ForkJoinTask 自体の依存関係は複雑ではなく、非同期タスク計算 FutureTask と同じように Future インターフェイスを実装します

    Java の Fork/Join フレームワークとメソッド呼び出し

    ForkJoinTask のコア ソース コードと、分割統治法によってタスクがどのように計算されるかを学習してみましょう。

    ForkJoinTask の中核は、fork() メソッドと join() メソッドです。

    fork()

    • 現在のスレッドが ForkJoinWorkerThread スレッドかどうかを確認します

      • #Yes 現在のスレッドをワーク キューに直接プッシュします。

      • ForkJoinPool の externalPush メソッドを呼び出すかどうか

    ##In

    ForkJoinPool 静的共通オブジェクトが構築され、ここで呼び出されるのは commonexternalPush()

    #join()## です。

      #doJoin() メソッドを呼び出し、スレッドの実行が完了するまで待ちます
    •     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 は、主に次の取得メソッドを実装する ForkJoinTask のサブクラスです。一般的な制約による結果。自分でタスクを作成する必要がある場合でも、RecursiveTask を実装し、コアの計算メソッド 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 スレッド プール

    ForkJoinTask の多くの関数は ForkJoinPool スレッド プールに依存しているため、ForkJoinTask は ForkJoinPool なしでは実行できません。ForkJoinPool は ThreadPoolExecutor と多くの類似点があります。これは特にスレッド プールに使用されます。 ForkJoinTask タスクを実行します。スレッド プール技術については以前記事で紹介しました。興味のある方は読んでください。スレッド プール (プーリング技術) の実装原理を Java ソース コードから分析します。

    ForkJoinPool ThreadPoolExecutor との継承関係はほぼ同じであり、兄弟に相当します。

    Java の Fork/Join フレームワークとメソッド呼び出しワークスチールアルゴリズム

    ワークスチールアルゴリズムはForkJoinPoolで採用されており、フォークサブタスクごとに新しいスレッドが作成されると、システムリソースに影響を与えます。オーバーヘッドが大きいため、スレッド プールを使用する必要があります。一般的なスレッド プールにはタスク キューが 1 つしかありませんが、ForkJoinPool の場合、同じタスクによってフォークされたサブタスクは並列であるため、効率を向上させ、スレッドの競合を減らすために、これらの並列タスクは別のキューに配置する必要があります。異なるタスクを異なる速度で処理すると、最初に独自のキュー内のタスクの実行を完了したスレッドが存在する可能性があります。このとき、効率を向上させるために、スレッドが他のスレッドを「

    スチール

    」することを許可できます。タスクキュー これはいわゆる「仕事盗みアルゴリズム」です。 一般的なキューの場合、キューに入る要素はキューの最後にあり、キューから出ていく要素は先頭にあります。「作業盗用」のニーズを満たすために、タスク キューはデキューをサポートする必要があります。 "tail" の要素を追加することで、他のワーカー スレッドとの競合を減らすことができます (他のワーカー スレッドは、チームの先頭から独自のタスク キュー内のタスクを取得するため)。この場合、両端のブロッキングを使用する必要があります。それを解決するためにキューに入れてください。

    構築メソッド

    まず、ForkJoinPool スレッド プールの構築メソッドを見てみましょう。これには 3 つの構築形式があり、その中で最も複雑なのは 4 つの入力パラメータの構築です。 4 つの入力パラメータが何を表しているか見てみましょう。

      int 並列処理 並列処理レベル (既存のスレッドの最大数を表すものではありません)
    • ForkJoinWorkerThreadFactory ファクトリ スレッド作成ファクトリ
    • UncaughtExceptionHandler ハンドラー例外キャプチャ ハンドラー
    • boolean asyncMode 先入れ先出し動作モードまたは後入れ先出し動作モード
    •     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();
          }
    • 提出方法

    タスクを提出する方法を見てみましょう:

    externalPush

    この方法は非常に便利です。私たちにとって馴染み深いのは、まさにフォーク時です。 現在のスレッドが ForkJoinWorkerThread でない場合、新しく送信されたタスクもこのメソッドを通じて実行されます。フォークとは、送信用の新しいサブタスクを作成することであることがわかります。 <p><code>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的模式,也可以按照具体需求进行灵活的设计。

    Java の Fork/Join フレームワークとメソッド呼び出し

    以上がJava の Fork/Join フレームワークとメソッド呼び出しの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。

    声明:
    この記事はyisu.comで複製されています。侵害がある場合は、admin@php.cn までご連絡ください。