浅谈Java Fork/Join并行框架

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
* Created by hujian06 on 2017/9/28.
*
* fork/join demo
*/
public class ForkJoinDemo {

 /**
  * how to find the max number in array by Fork/Join
  */
 private static class MaxNumber extends RecursiveTask<Integer> {

   private int threshold = 2;

   private int[] array; // the data array

   private int index0 = 0;
   private int index1 = 0;

   public MaxNumber(int[] array, int index0, int index1) {
     this.array = array;
     this.index0 = index0;
     this.index1 = index1;
   }

   @Override
   protected Integer compute() {
     int max = Integer.MIN_VALUE;
     if ((index1 - index0) <= threshold) {

       for (int i = index0;i <= index1; i ++) {
         max = Math.max(max, array[i]);
       }

     } else {
       //fork/join
       int mid = index0 + (index1 - index0) / 2;
       MaxNumber lMax = new MaxNumber(array, index0, mid);
       MaxNumber rMax = new MaxNumber(array, mid + 1, index1);

       lMax.fork();
       rMax.fork();

       int lm = lMax.join();
       int rm = rMax.join();

       max = Math.max(lm, rm);

     }

     return max;
   }
 }

 public static void main(String ... args) throws ExecutionException, InterruptedException, TimeoutException {

   ForkJoinPool pool = new ForkJoinPool();

   int[] array = {100,400,200,90,80,300,600,10,20,-10,30,2000,1000};

   MaxNumber task = new MaxNumber(array, 0, array.length - 1);

   Future<Integer> future = pool.submit(task);

   System.out.println("Result:" + future.get(1, TimeUnit.SECONDS));

 }

}

  volatile WorkQueue[] workQueues; 
  final ForkJoinWorkerThreadFactory factory;

  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();
  }
  
    final int doExec() {
    int s; boolean completed;
    if ((s = status) >= 0) {
      try {
        completed = exec();
      } catch (Throwable rex) {
        return setExceptionalCompletion(rex);
      }
      if (completed)
        s = setCompletion(NORMAL);
    }
    return s;
  }
  
   /**
   * Implements execution conventions for RecursiveTask.
   */
  protected final boolean exec() {
    result = compute();
    return true;
  }