fork/join全面剖析，你可以不用，但是不能不懂！

fork/join作为一个并发框架在jdk7的时候就加入到了我们的java并发包java.util.concurrent中，并且在java 8 的lambda并行流中充当着底层框架的角色。

在鄢陵等地区，都构建了全面的区域性战略布局，加强发展的系统性、市场前瞻性、产品创新能力，以专注、极致的服务理念，为客户提供网站设计制作、成都网站设计 网站设计制作按需求定制网站,公司网站建设,企业网站建设,成都品牌网站建设,营销型网站,成都外贸网站制作,鄢陵网站建设费用合理。

这样一个优秀的框架设计，我自己想了解一下它的底层代码是如何实现的，所以我尝试的去阅读了JDK相关的源码。下面我打算分享一下阅读完之后的心得~。

1、fork/join的设计思路

了解一个框架的第一件事，就是先了解别人的设计思路!

fork/join大体的执行过程就如上图所示，先把一个大任务分解(fork)成许多个独立的小任务，然后起多线程并行去处理这些小任务。处理完得到结果后再进行合并(join)就得到我们的最终结果。

显而易见的这个框架是借助了现代计算机多核的优势并行去处理数据。这看起来好像没有什么特别之处，这个套路很多人都会，并且工作中也会经常运用~。其实fork/join的最特别之处在于它还运用了一种叫work-stealing(工作窃取)的算法，这种算法的设计思路在于把分解出来的小任务放在多个双端队列中，而线程在队列的头和尾部都可获取任务。

当有线程把当前负责队列的任务处理完之后，它还可以从那些还没有处理完的队列的尾部窃取任务来处理，这连线程的空余时间也充分利用了！。

work-stealing原理图如下：

2、实现fork/join 定义了哪些角色？

了解设计原理，这仅仅是第一步！要了解别人整个的实现思路， 还需要了解别人为了实现这个框架定义了哪些角色，并了解这些角色的职责范围是什么的。因为知道谁负责了什么，谁做什么，这样整个逻辑才能串起来！在JAVA里面角色是以类的形式定义的,而了解类的行为最直接的方式就是看定义的公共方法~。

这里介绍JDK里面与fork/join相关的主要几个类：

  •  ForkJoinPool：充当fork/join框架里面的管理者，最原始的任务都要交给它才能处理。它负责控制整个fork/join有多少个workerThread，workerThread的创建，激活都是由它来掌控。它还负责workQueue队列的创建和分配，每当创建一个workerThread，它负责分配相应的workQueue。然后它把接到的活都交给workerThread去处理，它可以说是整个frok/join的容器。
  •  ForkJoinWorkerThread：fork/join里面真正干活的"工人"，本质是一个线程。里面有一个ForkJoinPool.WorkQueue的队列存放着它要干的活，接活之前它要向ForkJoinPool注册(registerWorker)，拿到相应的workQueue。然后就从workQueue里面拿任务出来处理。它是依附于ForkJoinPool而存活，如果ForkJoinPool的销毁了,它也会跟着结束。
  •  ForkJoinPool.WorkQueue: 双端队列就是它，它负责存储接收的任务。
  •  ForkJoinTask：代表fork/join里面任务类型，我们一般用它的两个子类RecursiveTask、RecursiveAction。这两个区别在于RecursiveTask任务是有返回值，RecursiveAction没有返回值。任务的处理逻辑包括任务的切分都集中在compute()方法里面。

3、fork/join初始化时做了什么

大到一个系统，小到一个框架，初始化工作往往是体现逻辑的一个重要地方！因为这是开始的地方，后面的逻辑会有依赖！所以把初始化看明白了，后面很多逻辑就容易理解多了。

下面上一段代码，(ps:这段代码是在网上找到的，并做了一小部分的修改) 

 
 
 
 
    
  
  
  
  
  1. public class CountTask extends RecursiveTask {  
    2. 
  
  
  
  
  2.     private static final int THRESHOLD = 2; //阀值  
    3. 
  
  
  
  
  3.     private int start;  
    4. 
  
  
  
  
  4.     private int end;  
    5. 
  
  
  
  
  5.     public CountTask(int start,int end){  
    6. 
  
  
  
  
  6.         this.start = start;  
    7. 
  
  
  
  
  7.         this.end = end;  
    8. 
  
  
  
  
  8.     }  
    9. 
  
  
  
  
  9.     @Override  
    10. 
  
  
  
  
  10.     protected Integer compute() {  
    11. 
  
  
  
  
  11.          int sum = 0;  
    12. 
  
  
  
  
  12.          boolean canCompute = (end - start) <= THRESHOLD;  
    13. 
  
  
  
  
  13.          if(canCompute){  
    14. 
  
  
  
  
  14.              for(int i = start; i <= end; i++){  
    15. 
  
  
  
  
  15.                  sum += i;  
    16. 
  
  
  
  
  16.              }  
    17. 
  
  
  
  
  17.          }else{  
    18. 
  
  
  
  
  18.              int middle = (start + end) / 2;  
    19. 
  
  
  
  
  19.              CountTask leftTask = new CountTask(start,middle);  
    20. 
  
  
  
  
  20.              CountTask rightTask = new CountTask(middle + 1,end);  
    21. 
  
  
  
  
  21.              //执行子任务  
    22. 
  
  
  
  
  22.              leftTask.fork();  
    23. 
  
  
  
  
  23.              rightTask.fork();  
    24. 
  
  
  
  
  24.              //等待子任务执行完，并得到其结果  
    25. 
  
  
  
  
  25.              Integer rightResult = rightTask.join();  
    26. 
  
  
  
  
  26.              Integer leftResult = leftTask.join();  
    27. 
  
  
  
  
  27.              //合并子任务  
    28. 
  
  
  
  
  28.              sum = leftResult + rightResult;  
    29. 
  
  
  
  
  29.          }  
    30. 
  
  
  
  
  30.          return sum;  
    31. 
  
  
  
  
  31.     }  
    32. 
  
  
  
  
  32.     public static void main(String[] args) throws ExecutionException, InterruptedException {  
    33. 
  
  
  
  
  33.         ForkJoinPool forkJoinPool = new ForkJoinPool();  
    34. 
  
  
  
  
  34.         CountTask countTask = new CountTask(1,200);  
    35. 
  
  
  
  
  35.         ForkJoinTask forkJoinTask = forkJoinPool.submit(countTask);  
    36. 
  
  
  
  
  36.         System.out.println(forkJoinTask.get());  
    37. 
  
  
  
  
  37.     }  
    38. 
  
  
  
  
  38. } 
    39.

代码的执行过程解释起来也是很简单就是把[1,200]，分成[1,100],[101,200],然后再对每个部分进行一个递归分解最终分解成[1,2],[3,4],[5,6]…..[199,200]独立的小任务，然后两两求和合并。

其实显然易见负责整个fork/join初始化工作的就是ForkJoinPool！初始化代码就是那一行 ForkJoinPool forkJoinPool = new ForkJoinPool(),点进去查看源码。 

 
 
 
 
    
  
  
  
  
  1. ForkJoinPool forkJoinPool = new ForkJoinPool();  
    2. 
  
  
  
  
  2. //最终调用到这段代码  
    3. 
  
  
  
  
  3. public ForkJoinPool(int parallelism,  
    4. 
  
  
  
  
  4.                     ForkJoinWorkerThreadFactory factory,  
    5. 
  
  
  
  
  5.                     UncaughtExceptionHandler handler,  
    6. 
  
  
  
  
  6.                     boolean asyncMode) {  
    7. 
  
  
  
  
  7.     this(checkParallelism(parallelism), //并行度,当前机器的cpu核数  
    8. 
  
  
  
  
  8.             checkFactory(factory), //工作线程创建工厂  
    9. 
  
  
  
  
  9.             handler, //异常处理handler  
    10. 
  
  
  
  
  10.             asyncMode ? FIFO_QUEUE : LIFO_QUEUE, //任务队列出队模式 异步：先进先出，同步：后进先出  
    11. 
  
  
  
  
  11.             "ForkJoinPool-" + nextPoolId() + "-worker-");  
    12. 
  
  
  
  
  12.     checkPermission();  
    13. 
  
  
  
  
  13. } 
    14.

看完初始化的代码我们可以知道原来创建ForkJoinPool创建workerThread的工作都是统一由一个叫ForkJoinWorkerThreadFactory的工厂去创建，创建出来的线程都有一个统一的前辍名称"ForkJoinPool-" + nextPoolId() + "-worker-".队列出队模式是LIFO(后进先出)，那这样后面的入队的任务是会被先处理的。

所以上面提到对代码做了一些修改就是先处理rightTask，再处理leftTask。这其实是对代码的一种优化！ 

 
 
 
 
    
  
  
  
  
  1. //执行子任务  
    2. 
  
  
  
  
  2.  leftTask.fork();  
    3. 
  
  
  
  
  3.  rightTask.fork();  
    4. 
  
  
  
  
  4.  Integer rightResult = rightTask.join();  
    5. 
  
  
  
  
  5.  Integer leftResult = leftTask.join(); 
    6.

4、任务的提交逻辑？

fork/join其实大部分逻辑处理操作都集中在提交任务和处理任务这两块，了解任务的提交基本上后面就很容易理解了。

fork/join提交任务主要分为两种：

第一种:第一次提交到forkJoinPool 

 
 
 
 
    
  
  
  
  
  1. ForkJoinTask forkJoinTask = forkJoinPool.submit(countTask); 
    2.

第二种:任务切分之后的提交 

 
 
 
 
    
  
  
  
  
  1. leftTask.fork();  
    2. 
  
  
  
  
  2. rightTask.fork(); 
    3.

提交到forkJoinPool :

代码调用路径 submit(ForkJoinTask task) -> externalPush(ForkJoinTask task) -> externalSubmit(ForkJoinTask task)

下面贴上externalSubmit的详细代码，着重留意注释的部分。 

 
 
 
 
    
  
  
  
  
  1. private void externalSubmit(ForkJoinTask task) {  
    2. 
  
  
  
  
  2.        int r;                                    // initialize caller's probe  
    3. 
  
  
  
  
  3.        if ((r = ThreadLocalRandom.getProbe()) == 0) {  
    4. 
  
  
  
  
  4.            ThreadLocalRandom.localInit();  
    5. 
  
  
  
  
  5.            r = ThreadLocalRandom.getProbe();  
    6. 
  
  
  
  
  6.        }  
    7. 
  
  
  
  
  7.        for (;;) { //采用循环入队的方式  
    8. 
  
  
  
  
  8.            WorkQueue[] ws; WorkQueue q; int rs, m, k;  
    9. 
  
  
  
  
  9.            boolean move = false;  
    10. 
  
  
  
  
  10.            if ((rs = runState) < 0) {  
    11. 
  
  
  
  
  11.                tryTerminate(false, false);     // help terminate  
    12. 
  
  
  
  
  12.                throw new RejectedExecutionException();  
    13. 
  
  
  
  
  13.            }  
    14. 
  
  
  
  
  14.            else if ((rs & STARTED) == 0 ||     // initialize 初始化操作  
    15. 
  
  
  
  
  15.                     ((ws = workQueues) == null || (m = ws.length - 1) < 0)) {  
    16. 
  
  
  
  
  16.                int ns = 0;  
    17. 
  
  
  
  
  17.                rs = lockRunState();  
    18. 
  
  
  
  
  18.                try {  
    19. 
  
  
  
  
  19.                    if ((rs & STARTED) == 0) {  
    20. 
  
  
  
  
  20.                        U.compareAndSwapObject(this, STEALCOUNTER, null,  
    21. 
  
  
  
  
  21.                                               new AtomicLong());  
    22. 
  
  
  
  
  22.                        // create workQueues array with size a power of two  
    23. 
  
  
  
  
  23.                        int p = config & SMASK; // ensure at least 2 slots //config就是cpu的核数  
    24. 
  
  
  
  
  24.                        int n = (p > 1) ? p - 1 : 1;  
    25. 
  
  
  
  
  25.                        n |= n >>> 1; n |= n >>> 2;  n |= n >>> 4;  
    26. 
  
  
  
  
  26.                        n |= n >>> 8; n |= n >>> 16; n = (n + 1) << 1; //算出workQueues的大小n,n一定是2的次方数  
    27. 
  
  
  
  
  27.                        workQueues = new WorkQueue[n];  //初始化队列，然后跳到最外面的循环继续把任务入队~  
    28. 
  
  
  
  
  28.                        ns = STARTED;  
    29. 
  
  
  
  
  29.                    }  
    30. 
  
  
  
  
  30.                } finally {  
    31. 
  
  
  
  
  31.                    unlockRunState(rs, (rs & ~RSLOCK) | ns);  
    32. 
  
  
  
  
  32.                }  
    33. 
  
  
  
  
  33.            }  
    34. 
  
  
  
  
  34.            else if ((q = ws[k = r & m & SQMASK]) != null) { //选中了一个一个非空队列  
    35. 
  
  
  
  
  35.                if (q.qlock == 0 && U.compareAndSwapInt(q, QLOCK, 0, 1)) { //利用cas操作加锁成功！  
    36. 
  
  
  
  
  36.                    ForkJoinTask[] a = q.array;  
    37. 
  
  
  
  
  37.                    int s = q.top;  
    38. 
  
  
  
  
  38.                    boolean submitted = false; // initial submission or resizing  
    39. 
  
  
  
  
  39.                    try {                      // locked version of push  
    40. 
  
  
  
  
  40.                        if ((a != null && a.length > s + 1 - q.base) ||  
    41. 
  
  
  
  
  41.                            (a = q.growArray()) != null) {  
    42. 
  
  
  
  
  42.                            int j = (((a.length - 1) & s) << ASHIFT) + ABASE; //计算出任务在队列中的位置  
    43. 
  
  
  
  
  43.                            U.putOrderedObject(a, j, task);  //把任务放在队列中  
    44. 
  
  
  
  
  44.                            U.putOrderedInt(q, QTOP, s + 1); //更新一次存放的位置  
    45. 
  
  
  
  
  45.                            submitted = true;  
    46. 
  
  
  
  
  46.                        }  
    47. 
  
  
  
  
  47.                    } finally {  
    48. 
  
  
  
  
  48.                        U.compareAndSwapInt(q, QLOCK, 1, 0); //利用cas操作释放锁！  
    49. 
  
  
  
  
  49.                    }  
    50. 
  
  
  
  
  50.                    if (submitted) {  
    51. 
  
  
  
  
  51.                        signalWork(ws, q);  
    52. 
  
  
  
  
  52.                        return; //任务入队成功了！跳出循环！  
    53. 
  
  
  
  
  53.                    }  
    54. 
  
  
  
  
  54.                }  
    55. 
  
  
  
  
  55.                move = true;                   // move on failure  
    56. 
  
  
  
  
  56.            }  
    57. 
  
  
  
  
  57.            else if (((rs = runState) & RSLOCK) == 0) { // create new queue 选中的队列是空，初始化完队列，然后继续入队！  
    58. 
  
  
  
  
  58.                q = new WorkQueue(this, null);  
    59. 
  
  
  
  
  59.                q.hint = r;  
    60. 
  
  
  
  
  60.                q.config = k | SHARED_QUEUE;  
    61. 
  
  
  
  
  61.                q.scanState = INACTIVE;  
    62. 
  
  
  
  
  62.                rs = lockRunState();           // publish index  
    63. 
  
  
  
  
  63.                if (rs > 0 &&  (ws = workQueues) != null &&  
    64. 
  
  
  
  
  64.                    k < ws.length && ws[k] == null)  
    65. 
  
  
  
  
  65.                    ws[k] = q;                 // else terminated  
    66. 
  
  
  
  
  66.                unlockRunState(rs, rs & ~RSLOCK);  
    67. 
  
  
  
  
  67.            }  
    68. 
  
  
  
  
  68.            else  
    69. 
  
  
  
  
  69.                move = true;                   // move if busy  
    70. 
  
  
  
  
  70.            if (move)  
    71. 
  
  
  
  
  71.                r = ThreadLocalRandom.advanceProbe(r);  
    72. 
  
  
  
  
  72.        }  
    73. 
  
  
  
  
  73.    } 
    74.

通过对externalSubmit方法的代码进行分析，我们知道了第一次提交任务给forkJoinPool时是在无限循环for (;;)中入队。第一步先检查workQueues是不是还没有创建，如果没有，则进行创建。之后跳到外层for循环并随机选取workQueues里面一个队列，并判断队列是否已创建。没有创建，则进行创建！后又跳到外层for循环直到选到一个非空队列并且加锁成功！这样最后才把任务入队~。

所以我们知道fork/join的任务队列workQueues并不是初始化的时候就创建好了，而是在有任务提交的时候才创建！并且每次入队时都需要利用cas操作来进行加锁和释放锁！

任务切分之后的提交： 

 
 
 
 
    
  
  
  
  
  1. public final ForkJoinTask fork() {  
    2. 
  
  
  
  
  2.       Thread t;  
    3. 
  
  
  
  
  3.       if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) 
    4. 
  
  
  
  
  4.           ((ForkJoinWorkerThread)t).workQueue.push(this); //workerThread直接入自己的workQueue  
    5. 
  
  
  
  
  5.       else  
    6. 
  
  
  
  
  6.           ForkJoinPool.common.externalPush(this);  
    7. 
  
  
  
  
  7.       return this;  
    8. 
  
  
  
  
  8.   }  
    9. 
 
 
 
 
 
 
 
 
    
  
  
  
  
  1. final void externalPush(ForkJoinTask task) {  
    2. 
  
  
  
  
  2.         WorkQueue[] ws; WorkQueue q; int m;  
    3. 
  
  
  
  
  3.         int r = ThreadLocalRandom.getProbe();  
    4. 
  
  
  
  
  4.         int rs = runState;  
    5. 
  
  
  
  
  5.         if ((ws = workQueues) != null && (m = (ws.length - 1)) >= 0 &&  
    6. 
  
  
  
  
  6.             (q = ws[m & r & SQMASK]) != null && r != 0 && rs > 0 &&  
    7. 
  
  
  
  
  7.             U.compareAndSwapInt(q, QLOCK, 0, 1)) { //随机选取了一个非空队列，并且加锁成功！下面是普通的入队过程~  
    8. 
  
  
  
  
  8.             ForkJoinTask[] a; int am, n, s;  
    9. 
  
  
  
  
  9.             if ((a = q.array) != null &&  
    10. 
  
  
  
  
  10.                 (aam = a.length - 1) > (n = (s = q.top) - q.base)) {  
    11. 
  
  
  
  
  11.                 int j = ((am & s) << ASHIFT) + ABASE;  
    12. 
  
  
  
  
  12.                 U.putOrderedObject(a, j, task);  
    13. 
  
  
  
  
  13.                 U.putOrderedInt(q, QTOP, s + 1);  
    14. 
  
  
  
  
  14.                 U.putIntVolatile(q, QLOCK, 0); 
    15. 
  
  
  
  
  15.                  if (n <= 1)  
    16. 
  
  
  
  
  16.                     signalWork(ws, q);  
    17. 
  
  
  
  
  17.                 return; //结束方法  
    18. 
  
  
  
  
  18.             }  
    19. 
  
  
  
  
  19.             U.compareAndSwapInt(q, QLOCK, 1, 0); //一定要释放锁！  
    20. 
  
  
  
  
  20.         }
            //这个就是上面的externalSummit方法，逻辑是一样的~  
    21. 
  
  
  
  
  21.         externalSubmit(task);  
    22. 
  
  
  
  
  22.     } 
    23.

从代码中我们知道了提交一个fork任务的过程和第一次提交到forkJoinPool的过程是大同小异的。主要区分了提交任务的线程是不是workerThread，如果是，任务直接入workerThread当前的workQueue，不是则尝试选中一个workQueue q。如果q非空并且加锁成功则进行入队，否则执行与第一次任务提交到forkJoinPool差不多的逻辑~。

5、任务的消费

提交到任务的最终目的，是为了消费任务并最终获取到我们想要的结果。介绍任务消费之前我们先了解一个我们的任务ForkJoinTask有哪些关键属性和方法。 

 
 
 
 
    
  
  
  
  
  1. /** The run status of this task */  
    2. 
  
  
  
  
  2. volatile int status; // accessed directly by pool and workers  
    3. 
  
  
  
  
  3. static final int DONE_MASK   = 0xf0000000;  // mask out non-completion bits  
    4. 
  
  
  
  
  4. static final int NORMAL      = 0xf0000000;  // must be negative  
    5. 
  
  
  
  
  5. static final int CANCELLED   = 0xc0000000;  // must be < NORMAL  
    6. 
  
  
  
  
  6. static final int EXCEPTIONAL = 0x80000000;  // must be < CANCELLED  
    7. 
  
  
  
  
  7. static final int SIGNAL      = 0x00010000;  // must be >= 1 << 16  
    8. 
  
  
  
  
  8. static final int SMASK       = 0x0000ffff;  // short bits for tags  
    9. 
 
 
 
 
 
 
 
 
    
  
  
  
  
  1. final int doExec() { //任务的执行入口  
    2. 
  
  
  
  
  2.   int s; boolean completed;  
    3. 
  
  
  
  
  3.   if ((s = status) >= 0) {  
    4. 
  
  
  
  
  4.       try {  
    5. 
  
  
  
  
  5.           completed = exec();  
    6. 
  
  
  
  
  6.       } catch (Throwable rex) {  
    7. 
  
  
  
  
  7.           return setExceptionalCompletion(rex);  
    8. 
  
  
  
  
  8.       }  
    9. 
  
  
  
  
  9.       if (completed)  
    10. 
  
  
  
  
  10.           s = setCompletion(NORMAL);  
    11. 
  
  
  
  
  11.   }  
    12. 
  
  
  
  
  12.   return s;  
    13. 
  
  
  
  
  13.  } 
    14.

再看一下RecursiveTask的定义 

 
 
 
 
    
  
  
  
  
  1. public abstract class RecursiveTask extends ForkJoinTask {  
    2. 
  
  
  
  
  2.     private static final long serialVersionUID = 5232453952276485270L;  
    3. 
  
  
  
  
  3.     /**  
    4. 
  
  
  
  
  4.      * The result of the computation.  
    5. 
  
  
  
  
  5.      */  
    6. 
  
  
  
  
  6.     V result;  
    7. 
  
  
  
  
  7.     /**  
    8. 
  
  
  
  
  8.      * The main computation performed by this task.  
    9. 
  
  
  
  
  9.      * @return the result of the computation 
    10. 
  
  
  
  
  10.      */  
    11. 
  
  
  
  
  11.     protected abstract V compute(); //我们实现的处理逻辑  
    12. 
  
  
  
  
  12.     public final V getRawResult() { //获取返回计算结果  
    13. 
  
  
  
  
  13.         return result;  
    14. 
  
  
  
  
  14.     }  
    15. 
  
  
  
  
  15.     protected final void setRawResult(V value) {  
    16. 
  
  
  
  
  16.         result = value;  
    17. 
  
  
  
  
  17.     }  
    18. 
  
  
  
  
  18.     /**  
    19. 
  
  
  
  
  19.      * Implements execution conventions for RecursiveTask.  
    20. 
  
  
  
  
  20.      */  
    21. 
  
  
  
  
  21.     protected final boolean exec() {  
    22. 
  
  
  
  
  22.         result = compute(); //存储计算结果  
    23. 
  
  
  
  
  23.         return true;  
    24. 
  
  
  
  
  24.     }  
    25. 
  
  
  
  
  25. } 
    26.

在代码中我们看到任务的真正执行链路是 doExec -> exec -> compute -> 最后设置status 和 result。既然定义状态status并且还是volatile类型我们可以推断出workerThread在获取到执行任务之后都会先判断status是不是已完成或者异常状态，才决定要不要处理该任务。

下面看一下任务真正的处理逻辑代码！ 

 
 
 
 
    
  
  
  
  
  1. Integer rightResult = rightTask.join()  
    2. 
  
  
  
  
  2.   public final V join() {  
    3. 
  
  
  
  
  3.       int s;  
    4. 
  
  
  
  
  4.       if ((s = doJoin() & DONE_MASK) != NORMAL)  
    5. 
  
  
  
  
  5.           reportException(s);  
    6. 
  
  
  
  
  6.       return getRawResult();  
    7. 
  
  
  
  
  7.    }  
    8. 
  
  
  
  
  8.   //执行处理前先判断staus是不是已完成，如果完成了就直接返回  
    9. 
  
  
  
  
  9.  //因为这个任务可能被其它线程窃取过去处理完了  
    10. 
  
  
  
  
  10.   private int doJoin() {  
    11. 
  
  
  
  
  11.       int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w;  
    12. 
  
  
  
  
  12.       return (s = status) < 0 ? s :  
    13. 
  
  
  
  
  13.           ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?  
    14. 
  
  
  
  
  14.           (w = (wt = (ForkJoinWorkerThread)t).workQueue).  
    15. 
  
  
  
  
  15.           tryUnpush(this) && (s = doExec()) < 0 ? s :  
    16. 
  
  
  
  
  16.           wt.pool.awaitJoin(w, this, 0L) :  
    17. 
  
  
  
  
  17.           externalAwaitDone();  
    18. 
  
  
  
  
  18.   }  
    19.

代码的调用链是从上到下。整体处理逻辑如下：

线程是workerThread：

先判断任务是否已经处理完成，任务完成直接返回，没有则直接尝试出队tryUnpush(this) 然后执行任务处理doExec()。如果没有出队成功或者处理成功，则执行wt.pool.awaitJoin(w, this, 0L)。wt.pool.awaitJoin(w, this, 0L)的处理逻辑简单来说也是在一个for(;;）中不断的轮询任务的状态是不是已完成，完成就直接退出方法。否就继续尝试出队处理。直到任务完成或者超时为止。

线程不是workerThread:

直接进行入externalAwaitDone() 

 
 
 
 
    
  
  
  
  
  1. private int externalAwaitDone() {  
    2. 
  
  
  
  
  2.        int s = ((this instanceof CountedCompleter) ? // try helping  
    3. 
  
  
  
  
  3.                 ForkJoinPool.common.externalHelpComplete(  
    4. 
  
  
  
  
  4.                     (CountedCompleter)this, 0) :  
    5. 
  
  
  
  
  5.                 ForkJoinPool.common.tryExternalUnpush(this) ? doExec() : 0);  
    6. 
  
  
  
  
  6.        if (s >= 0 && (s = status) >= 0) {  
    7. 
  
  
  
  
  7.            boolean interrupted = false;  
    8. 
  
  
  
  
  8.            do {  
    9. 
  
  
  
  
  9.                if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {  
    10. 
  
  
  
  
  10.                    synchronized (this) {  
    11. 
  
  
  
  
  11.                        if (status >= 0) {  
    12. 
  
  
  
  
  12.                            try {  
    13. 
  
  
  
  
  13.                                wait(0L);  
    14. 
  
  
  
  
  14.                            } catch (InterruptedException ie) {  
    15. 
  
  
  
  
  15.                                interrupted = true;  
    16. 
  
  
  
  
  16.                            }  
    17. 
  
  
  
  
  17.                        } 
    18. 
  
  
  
  
  18.                        else  
    19. 
  
  
  
  
  19.                            notifyAll();  
    20. 
  
  
  
  
  20.                    }  
    21. 
  
  
  
  
  21.                }  
    22. 
  
  
  
  
  22.            } while ((s = status) >= 0);  
    23. 
  
  
  
  
  23.            if (interrupted)  
    24. 
  
  
  
  
  24.                Thread.currentThread().interrupt();  
    25. 
  
  
  
  
  25.        }  
    26. 
  
  
  
  
  26.        return s; 
    27.

externalAwaitDone的处理逻辑其实也比较简单，当前线程自己先尝试把任务出队ForkJoinPool.common.tryExternalUnpush(this) ? doExec()然后处理掉，如果不成功就交给workerThread去处理，然后利用object/wait的经典方法去监听任务status的状态变更。

6、任务的窃取

一直说fork/join的任务是work-stealing(工作窃取)，那任务究竟是怎么被窃取的呢。我们分析一下任务是由workThread来窃取的，workThread是一个线程。线程的所有逻辑都是由run()方法执行，所以任务的窃取逻辑一定在run()方法中可以找到！ 

 
 
 
 
    
  
  
  
  
  1. public void run() { //线程run方法  
    2. 
  
  
  
  
  2.        if (workQueue.array == null) { // only run once  
    3. 
  
  
  
  
  3.            Throwable exception = null;  
    4. 
  
  
  
  
  4.            try {  
    5. 
  
  
  
  
  5.                onStart();  
    6. 
  
  
  
  
  6.                pool.runWorker(workQueue);  //在这里处理任务队列！  
    7. 
  
  
  
  
  7.            } catch (Throwable ex) {  
    8. 
  
  
  
  
  8.                exexception = ex; 
    9. 
  
  
  
  
  9.            } finally {  
    10. 
  
  
  
  
  10.                try {  
    11. 
  
  
  
  
  11.                    onTermination(exception);  
    12. 
  
  
  
  
  12.                } catch (Throwable ex) {  
    13. 
  
  
  
  
  13.                    if (exception == null)  
    14. 
  
  
  
  
  14.                        exexception = ex;  
    15. 
  
  
  
  
  15.                } finally {  
    16. 
  
  
  
  
  16.                    pool.deregisterWorker(this, exception);  
    17. 
  
  
  
  
  17.                }  
    18. 
  
  
  
  
  18.            }  
    19. 
  
  
  
  
  19.        }  
    20. 
  
  
  
  
  20.    }  
    21. 
  
  
  
  
  21.   /**  
    22. 
  
  
  
  
  22.     * Top-level runloop for workers, called by ForkJoinWorkerThread.run.  
    23. 
  
  
  
  
  23.     */  
    24. 
  
  
  
  
  24.    final void runWorker(WorkQueue w) {  
    25. 
  
  
  
  
  25.        w.growArray();                   // allocate queue  进行队列的初始化  
    26. 
  
  
  
  
  26.        int seed = w.hint;               // initially holds randomization hint  
    27. 
  
  
  
  
  27.        int r = (seed == 0) ? 1 : seed;  // avoid 0 for xorShift  
    28. 
  
  
  
  
  28.        for (ForkJoinTask t;;) { //又是无限循环处理任务！  
    29. 
  
  
  
  
  29.            if ((t = scan(w, r)) != null) //在这里获取任务！  
    30. 
  
  
  
  
  30.                w.runTask(t);  
    31. 
  
  
  
  
  31.            else if (!awaitWork(w, r))  
    32. 
  
  
  
  
  32.                break;  
    33. 
  
  
  
  
  33.            r ^= r << 13; r ^= r >>> 17; r ^= r << 5; // xorshift  
    34. 
  
             
    
                分享标题：fork/join全面剖析，你可以不用，但是不能不懂！
                
    
                路径分享：http://www.shufengxianlan.com/qtweb/news15/247715.html
            
    
            
    
                网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等
                                
    
            
    
                广告
                
            
    
            
    
                声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源：
                创新互联

猜你还喜欢下面的内容

网站策划知识

同城分类信息

网站解决方案　　 　平山县网站建设　　 　成都ssl证书认证　　 　堡垒主机　　 　做网站的公司　　 　域名注册　　 　云南护栏打桩机　　 　成都托管服务器　　 　成都做网站　　 　南充电信机房托管　　 　虚拟主机　　 　眉山联通机房　　 　德阳服务器托管　　 　Google外贸推广　　 　中晴建设　　 　乐山网站建设　　 　成都租用服务器　　 　自适应网站　　 　域名申请　　 　服务器机柜租用