ForkJoinPool線程池

1.   使用場景： 對1000000000000000000000000求和

 

       ForkJoinPool 使用時，先將任務 拆分 成 小任務 和 合並任務兩部分

public class RecursiveActionTest extends RecursiveTask<Integer> {
    private static final long serialVersionUID = -3611254198265061729L;
    //閥值
    public static final int threshold = 10;
    private int start;
    private int end;
    public RecursiveActionTest(int start, int end) {
        this.start = start;
        this.end = end;
    }

    //compute 是一個遞歸任務
    @Override
    protected Integer compute() {
        int sum = 0;

        //如果任務足夠小就計算任務
        boolean canCompute = (end - start) <= threshold;
        if (canCompute) {    //執行加法任務
            for (int i = start; i <= end; i++) {
                // 每個任務為：sum+i( 此處添加任務)
                sum += i;
            }
        } else {
            // 如果任務大於閾值，就分裂成10子任務計算,step：每一份的任務數
            int step = (start + end) / 10;
            ArrayList<RecursiveActionTest> subTaskList = new ArrayList<RecursiveActionTest>();
            int pos = start;
            //設置每個小任務的始起始值和終止值

            for (int i = 0; i < 10; i++) {
                int lastOne = pos + step;
                if (lastOne > end) lastOne = end;

                RecursiveActionTest subTask = new RecursiveActionTest(pos, lastOne);
                pos += step + 1;
                subTaskList.add(subTask);
                //把子任務推向線程池
                subTask.fork();
            }
            // 等待所有子任務完成，並計算值
            for (RecursiveActionTest task : subTaskList) {
                sum += task.join();
            }
        }

      return sum;
    }

    public static void main(String[] args) {
        ForkJoinPool forkjoinPool = new ForkJoinPool();

        //生成一個計算任務，計算1+2+3+4
        RecursiveActionTest task = new RecursiveActionTest(1, 1000000000000000000000000);

        //執行一個任務
        Future<Integer> result = forkjoinPool.submit(task);

        try {
            System.out.println(result.get());
        } catch (Exception e) {
            System.out.println(e);
        }
    }

}

 

 

 

2.   forkJoinPool使用步驟：

 

3. ForkJoinPool 使用場景是什么？

 答： 比如要對1000萬個數據進行排序，那么會將這個任務分割成兩個500萬的排序任務和一個針對這兩組500萬數據的合並任務。以此類推，對於500萬的數據也會做出同樣的分割處理，到最后會設置一個閾值來規定當數據規模到多少時，停止這樣的分割處理。比如，當元素的數量小於10時，會停止分割，轉而使用插入排序對它們進行排序。

 

4. ForkJoinPool另外一個使用例子

public class MakeMoneyTask extends RecursiveTask<Integer>{
  private static final int MIN_GOAL_MONEY = 100000;
  private int goalMoney;
  private String name;
  private static final AtomicLong employeeNo = new AtomicLong();
  public MakeMoneyTask(int goalMoney){
    this.goalMoney = goalMoney;
    this.name = "員工" + employeeNo.getAndIncrement() + "號";
  }
  @Override
  protected Integer compute() {
    if (this.goalMoney < MIN_GOAL_MONEY){
      System.out.println(name + ": 老板交代了,要賺 " + goalMoney + " 元,為了買車買房,加油吧....");
      return makeMoney();
    }else{
      int subThreadCount = ThreadLocalRandom.current().nextInt(10) + 2;
      System.out.println(name + ": 上級要我賺 " + goalMoney + ", 有點小多,沒事讓我" + subThreadCount + "個手下去完成吧," +
          "每人賺個 " + Math.ceil(goalMoney * 1.0 / subThreadCount) + "元應該沒問題...");
      List<MakeMoneyTask> tasks = new ArrayList<>();
      for (int i = 0; i < subThreadCount; i ++){
        tasks.add(new MakeMoneyTask(goalMoney / subThreadCount));
      }
      Collection<MakeMoneyTask> makeMoneyTasks = invokeAll(tasks);
      int sum = 0;
      for (MakeMoneyTask moneyTask : makeMoneyTasks){
        try {
          sum += moneyTask.get();
        } catch (Exception e) {
          e.printStackTrace();
        }
      }
      System.out.println(name + ": 嗯,不錯,效率還可以,終於賺到 " + sum + "元,趕緊邀功去....");
      return sum;
    }
  }
  private Integer makeMoney(){
    int sum = 0;
    int day = 1;
    try {
      while (true){
        Thread.sleep(ThreadLocalRandom.current().nextInt(500));
        int money = ThreadLocalRandom.current().nextInt(MIN_GOAL_MONEY / 3);
        System.out.println(name + ": 在第 " + (day ++) + " 天賺了" + money);
        sum += money;
        if (sum >= goalMoney){
          System.out.println(name + ": 終於賺到 " + sum + " 元, 可以交差了...");
          break;
        }
      }
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    return sum;
  }
}