jdk-Futeure&ForkJoin框架源码学习

ForkJoinTask 主要借助了 Unsafe 类的 CAS 操作实现了对各项任务状态的更新。在继续执行完成各项任务时，调用 onCompletion(CountedCompleter caller) 形式通知该各项任务的依赖各项任务或向等待该各项任务的缓存发送信号，以此实现对各项任务结论的分拆和传递。

属性：

REOPR_SIGNAL: 表示各项任务的初始状态，即等待被继续执行。

 DONE_MASK: 各项任务完成状态的标识。

 NORMAL: 各项任务正常完成。

CANCELLED: 各项任务被取消。

 EXCEPTIONAL: 各项任务发生异常。

status: 表示各项任务的状态，取值可能为 REOPR_SIGNAL、NORMAL、CANCELLED 或 EXCEPTIONAL 中的两个。

 ForkJoinTask next: 指向下两个等待继续执行的各项任务。

Thread runner: 继续执行该各项任务的缓存。

 statusFlags: 表示各项任务的状态及其他一些控制信息。

 mode: 表示各项任务的运行模式。

Throwable exception: 各项任务继续执行过程中发生异常时，保存该异常对象。

形式：

 java.util.concurrent;

 java.io.Serializable;

java.util.Collection;

 java.util.List;

 java.util.RandomAccess;

 java.lang.ref.WeakReference;

 java.lang.ref.ReferenceQueue;

 java.util.concurrent.Callable;

java.util.concurrent.CancellationException;

 java.util.concurrent.ExecutionException;

 java.util.concurrent.Future;

 java.util.concurrent.RejectedExecutionException;

 java.util.concurrent.RunnableFuture;

java.util.concurrent.TimeUnit;

 java.util.concurrent.TimeoutException;

java.util.concurrent.locks.ReentrantLock;

 java.lang.reflect.Constructor;

{

{

 s;;) {

)

 s;

, STATUS, s, s | completion)) {

)

) { notifyAll(); }

 completion;

            }

        }

    }

{

completed;

) {

 {

                completed = exec();

 (Throwable rex) {

setExceptionalCompletion(rex);

            }

 (completed)

                s = setCompletion(NORMAL);

        }

 s;

    }

{

 s;

, STATUS, s, s | SIGNAL)) {

) {

)

 (InterruptedException ie) { }

                    notifyAll();

            }

        }

    }

{

ForkJoinPool.common.externalHelpComplete(

) :

);

) {

;

 {

, STATUS, s, s | SIGNAL)) {

) {

) {

 {

);

 (InterruptedException ie) {

;

                            }

                        }

                            notifyAll();

                    }

                }

);

 (interrupted)

                Thread.currentThread().interrupt();

        }

 s;

    }

{

 s;

(Thread.interrupted())

 InterruptedException();

 &&

 CountedCompleter) ?

                  ForkJoinPool.common.externalHelpComplete(

) :

) ? doExec() :

) {

) {

, STATUS, s, s | SIGNAL)) {

) {

)

);

                            notifyAll();

                    }

                }

            }

        }

 s;

    }

{

s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w;

 ? s :

 ForkJoinWorkerThread) ?

            (w = (wt = (ForkJoinWorkerThread)t).workQueue).

 ? s :

) :

            externalAwaitDone();

    }

{

 s; Thread t; ForkJoinWorkerThread wt;

 ? s :

 ForkJoinWorkerThread) ?

(wt = (ForkJoinWorkerThread)t).pool.

) :

            externalAwaitDone();

    }

 ExceptionNode[] exceptionTable;

ReentrantLock exceptionTableLock;

 ReferenceQueue

;

{

 Throwable ex;

        ExceptionNode next;

        ExceptionNode(ForkJoinTask task, Throwable ex, ExceptionNode next) {

(task, exceptionTableRefQueue);

.ex = ex;

.next = next;

.thrower = Thread.currentThread().getId();

.hashCode = System.identityHashCode(task);

        }

    }

{

 s;

) {

);

ReentrantLock lock = exceptionTableLock;

            lock.lock();

 {

                expungeStaleExceptions();

                ExceptionNode[] t = exceptionTable;

);

 (ExceptionNode e = t[i]; ; e = e.next) {

) {

, ex, t[i]);

;

                    }

;

                }

 {

                lock.unlock();

            }

s = setCompletion(EXCEPTIONAL);

        }

 s;

    }

{

 s = recordExceptionalCompletion(ex);

 ((s & DONE_MASK) == EXCEPTIONAL)

internalPropagateException(ex);

 s;

    }

{

    }

{

) {

 {

);

 (Throwable ignore) {

}

        }

    }

{

);

 ReentrantLock lock = exceptionTableLock;

        lock.lock();

 {

            ExceptionNode[] t = exceptionTable;

);

            ExceptionNode e = t[i];

;

) {

ExceptionNode next = e.next;

) {

)

                        t[i] = next;

                        pred.next = next;

;

                }

pred = e;

                e = next;

            }

            expungeStaleExceptions();

;

 {

            lock.unlock();

        }

    }

{

 ((status & DONE_MASK) != EXCEPTIONAL)

;

);

        ExceptionNode e;

ReentrantLock lock = exceptionTableLock;

        lock.lock();

 {

            expungeStaleExceptions();

            ExceptionNode[] t = exceptionTable;

)];

)

                e = e.next;

 {

            lock.unlock();

        }

        Throwable ex;

)

;

 (e.thrower != Thread.currentThread().getId()) {

            Class ec = ex.getClass();

 {

;

; i < cs.length; ++i) {

                    Constructor c = cs[i];

                    Class [] ps = c.getParameterTypes();

)

                        noArgCtor = c;

] == Throwable.class) {

Throwable wx = (Throwable)c.newInstance(ex);

) ? ex : wx;

                    }

                }

) {

Throwable wx = (Throwable)(noArgCtor.newInstance());

) {

                        wx.initCause(ex);

 wx;

                    }

                }

 (Exception ignore) {

            }

        }

 ex;

    }

{

😉 {

 ExceptionNode) {

hashCode = ((ExceptionNode)x).hashCode;

                ExceptionNode[] t = exceptionTable;

);

                ExceptionNode e = t[i];

;

) {

                    ExceptionNode next = e.next;

 (e == x) {

)

t[i] = next;

                            pred.next = next;

;

                    }

                    pred = e;

                    e = next;

                }

            }

        }

    }

{

 ReentrantLock lock = exceptionTableLock;

 (lock.tryLock()) {

 {

expungeStaleExceptions();

 {

                lock.unlock();

            }

        }

    }

{

)

            ForkJoinTask.uncheckedThrow(ex);

    }

{

    }

{

 (s == CANCELLED)

 CancellationException();

(s == EXCEPTIONAL)

            rethrow(getThrowableException());

    }

{

        Thread t;

 ForkJoinWorkerThread)

);

);

;

    }

{

 s;

 ((s = doJoin() & DONE_MASK) != NORMAL)

            reportException(s);

 getRawResult();

    }

{

 s;

((s = doInvoke() & DONE_MASK) != NORMAL)

            reportException(s);

 getRawResult();

    }

{

 s1, s2;

        t2.fork();

((s1 = t1.doInvoke() & DONE_MASK) != NORMAL)

            t1.reportException(s1);

 ((s2 = t2.doJoin() & DONE_MASK) != NORMAL)

            t2.reportException(s2);

    }

{

;

;

; –i) {

            ForkJoinTask t = tasks[i];

) {

)

 NullPointerException();

            }

)

t.fork();

)

ex = t.getException();

        }

; i <= last; ++i) {

            ForkJoinTask t = tasks[i];

) {

)

);

 (t.doJoin() < NORMAL)

                    ex = t.getException();

            }

        }

)

rethrow(ex);

    }

{

 List )) {

 ForkJoinTask [tasks.size()]));

 tasks;

        }

)

        List > ts =

            (List >) tasks;

;

;

; –i) {

            ForkJoinTask t = ts.get(i);

) {

)

 NullPointerException();

            }

)

                t.fork();

)

ex = t.getException();

        }

; i <= last; ++i) {

            ForkJoinTask t = ts.get(i);

) {

)

);

 (t.doJoin() < NORMAL)

                    ex = t.getException();

            }

        }

)

            rethrow(ex);

 tasks;

    }

{

(setCompletion(CANCELLED) & DONE_MASK) == CANCELLED;

    }

{

;

    }

{

 (status & DONE_MASK) == CANCELLED;

    }

{

 status < NORMAL;

    }

{

 (status & DONE_MASK) == NORMAL;

    }

{

RuntimeException) ||

 Error) ? ex :

 RuntimeException(ex));

    }

{

 {

            setRawResult(value);

 (Throwable rex) {

            setExceptionalCompletion(rex);

;

        }

setCompletion(NORMAL);

    }

{

        setCompletion(NORMAL);

}

{

 ForkJoinWorkerThread) ?

            doJoin() : externalInterruptibleAwaitDone();

Throwable ex;

 ((s &= DONE_MASK) == CANCELLED)

 CancellationException();

)

 ExecutionException(ex);

 getRawResult();

    }

{

 s;

 nanos = unit.toNanos(timeout);

 (Thread.interrupted())

InterruptedException();

) {

 d = System.nanoTime() + nanos;

            Thread t = Thread.currentThread();

 ForkJoinWorkerThread) {

ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t;

, deadline);

            }

 CountedCompleter) ?

                           ForkJoinPool.common.externalHelpComplete(

) :

) ?

) {

 &&

) {

 &&

, STATUS, s, s | SIGNAL)) {

) {

)

                                notifyAll();

                        }

                    }

                }

            }

        }

)

            s = status;

 ((s &= DONE_MASK) != NORMAL) {

            Throwable ex;

 (s == CANCELLED)

CancellationException();

 (s != EXCEPTIONAL)

 TimeoutException();

)

 ExecutionException(ex);

        }

 getRawResult();

    }

{

        doJoin();

    }

{

        doInvoke();

    }

{

        Thread t;

 ForkJoinWorkerThread) {

            ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t;

wt.pool.helpQuiescePool(wt.workQueue);

        }

            ForkJoinPool.quiesceCommonPool();

    }

{

 ((status & DONE_MASK) == EXCEPTIONAL)

            clearExceptionalCompletion();

;

    }

{

        Thread t = Thread.currentThread();

ForkJoinWorkerThread) ?

;

    }

{

 ForkJoinWorkerThread;

    }

{

        Thread t;

ForkJoinWorkerThread) ?

) :

));

    }

{

Thread t; ForkJoinPool.WorkQueue q;

 ForkJoinWorkerThread)

q = ((ForkJoinWorkerThread)t).workQueue;

            q = ForkJoinPool.commonSubmitterQueue();

 : q.queueSize();

    }

{

 ForkJoinPool.getSurplusQueuedTaskCount();

    }

;

;

;

 ForkJoinTask peekNextLocalTask() {

        Thread t; ForkJoinPool.WorkQueue q;

 ForkJoinWorkerThread)

            q = ((ForkJoinWorkerThread)t).workQueue;

q = ForkJoinPool.commonSubmitterQueue();

 : q.peek();

    }

 ForkJoinTask pollNextLocalTask() {

        Thread t;

ForkJoinWorkerThread) ?

            ((ForkJoinWorkerThread)t).workQueue.nextLocalTask() :

;

    }

 ForkJoinTask pollTask() {

        Thread t; ForkJoinWorkerThread wt;

 ForkJoinWorkerThread) ?

(wt = (ForkJoinWorkerThread)t).pool.nextTaskFor(wt.workQueue) :

;

    }

{

)status;

    }

{

s;;) {

, STATUS, s = status,

                                    (s & ~SMASK) | (tag & SMASK)))

)s;

        }

    }

{

s;;) {

)(s = status) != e)

;

, STATUS, s,

(s & ~SMASK) | (tag & SMASK)))

;

        }

    }

{

 Runnable runnable;

        T result;

AdaptedRunnable(Runnable runnable, T result) {

 NullPointerException();

.runnable = runnable;

        }

 result; }

{ result = v; }

; }

{ invoke(); }

;

    }

{

 Runnable runnable;

AdaptedRunnableAction(Runnable runnable) {

 NullPointerException();

.runnable = runnable;

        }

; }

{ }

; }

{ invoke(); }

;

    }

{

 Runnable runnable;

        RunnableExecuteAction(Runnable runnable) {

 NullPointerException();

.runnable = runnable;

        }

; }

{ }

; }

{

        }

;

    }

{

 Callable callable;

        T result;

AdaptedCallable(Callable callable) {

 NullPointerException();

.callable = callable;

        }

 result; }

{ result = v; }

{

 {

                result = callable.call();

;

 (Error err) {

 err;

(RuntimeException rex) {

 rex;

 (Exception ex) {

 RuntimeException(ex);

            }

        }

{ invoke(); }

;

    }

 ForkJoinTask adapt(Runnable runnable) {

 AdaptedRunnableAction(runnable);

    }

{

AdaptedRunnable(runnable, result);

    }

{

 AdaptedCallable(callable);

    }

;

{

        s.defaultWriteObject();

        s.writeObject(getException());

    }

{

s.defaultReadObject();

        Object ex = s.readObject();

)

setExceptionalCompletion((Throwable)ex);

    }

 sun.misc.Unsafe U;

 STATUS;

 {

 ReentrantLock();

 ReferenceQueue

 ExceptionNode[EXCEPTION_MAP_CAPACITY];

 {

U = sun.misc.Unsafe.getUnsafe();

            Class k = ForkJoinTask.class;

            STATUS = U.objectFieldOffset

));

 (Exception e) {

 Error(e);

        }

    }

最后

非常可惜到目前为止，我在很多项目中及跟很多做开发的同事或朋友，这个ForkJoin用到的很少，绝大部分还是停留 在CURD，好可惜，但是其实数据库查询强化、图像处置、机器自学、大数据处置等那些情景都能通过ForkJoin的博戈达计算能力来实现那些情景。当然这个ForkJoin还有非常非常多的用法，能另行了解。