title: 2018-8-3 聊聊 fork-join框架
tags: java骂倘，并發(fā)娶耍，線程池
grammar_cjkRuby: true

what闭树？

Fork/Join框架是Java7提供了的一個用于并行執(zhí)行任務(wù)的框架挥萌， 是一個把大任務(wù)分割成若干個小任務(wù)悍引，最終匯總每個小任務(wù)結(jié)果后得到大任務(wù)結(jié)果的框架家肯。

why?

同樣是線程池倦挂，為什么我們要用Fork/Join不用別的棋枕。

工作竊取算法（work-stealing）:

先執(zhí)行完當前線程的任務(wù)，直接去從其他為執(zhí)行完任務(wù)的線程中竊取任務(wù)執(zhí)行妒峦；

圖：

任務(wù)竊取

如圖線程1先執(zhí)行完自己的4個任務(wù)重斑，線程2還有沒執(zhí)行的任務(wù)，線程1接著從線程2中獲取未執(zhí)行的任務(wù)肯骇。實現(xiàn)是基于 雙向隊列

How窥浪？

RecursiveAction

執(zhí)行沒有返回結(jié)果的任務(wù)

• 新建TestRecursiveAction 對任務(wù)進行分割的主線程類

package com.fengxg.test.recursive.action;

import java.util.concurrent.RecursiveAction;

/**
 * @Desccription 用于沒有返回結(jié)果的任務(wù)，對任務(wù)進行分割的主線程
 * @auther Fengxg
 * @create 2018/7/25
 */
public class TestRecursiveAction extends RecursiveAction {

    private static final int ONE_TASK_DEAL_NUMBER = 10;

    /**
     * 計數(shù)器笛丙，計算器執(zhí)行完漾脂，表示最后一個任務(wù)也跑完
     */

    String[] data = {};

    public TestRecursiveAction(String[] data) {
        super();
        this.data = data;
    }

    @Override
    protected void compute() {
        //需要拆分成多少個任務(wù)單數(shù)
        int needSplitCount = 1;
        //剩余多少任務(wù)單
        int leftCount = data.length % ONE_TASK_DEAL_NUMBER;
        needSplitCount = (leftCount == 0?data.length/ONE_TASK_DEAL_NUMBER:data.length/ONE_TASK_DEAL_NUMBER+1);
        //初始化計數(shù)器
//        countDownLatch = new CountDownLatch(needSplitCount);

        //拆分任務(wù)單
        for(int i=0; i<needSplitCount; i++)
        {
            ActionWorkTask workTask = null;
            if(i == needSplitCount-1 && leftCount!=0)
            {
                workTask = new ActionWorkTask(i*ONE_TASK_DEAL_NUMBER, i*ONE_TASK_DEAL_NUMBER+leftCount, data,i);
            }else {
                workTask = new ActionWorkTask(i*ONE_TASK_DEAL_NUMBER, (i+1)*ONE_TASK_DEAL_NUMBER, data,i);
            }
            //提交到線程池隊列中
            workTask.fork();
        }
    }

    public String[] getData() {
        return data;
    }

    public void setData(String[] data) {
        this.data = data;
    }
}

• 新建ActionWorkTask 分片后執(zhí)行具體業(yè)務(wù)任務(wù)的類

package com.fengxg.test.recursive.action;

import java.util.concurrent.RecursiveAction;

/**
 * @Desccription 分片后執(zhí)行業(yè)務(wù)任務(wù)的具體類
 * @auther Fengxg
 * @create 2018/7/25
 */
public class ActionWorkTask  extends RecursiveAction{

    public ActionWorkTask(int startIndex, int endIndex, String[] datas,int workNum) {
        super();
        this.startIndex = startIndex;
        this.endIndex = endIndex;
        this.datas = datas;
        this.workNum = workNum;
    }

    private int workNum;
    private int startIndex;
    private int endIndex;
//    private CountDownLatch countDownLatch;
    private String[] datas;

    @Override
    protected void compute() {
        Thread t = Thread.currentThread();
        for (int i = startIndex; i < endIndex; i++) {
            try {
                Thread.sleep(100L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("workNum" + workNum  +"，第" + datas[i] + "任務(wù)," + "執(zhí)行線程:" + t.getName());
        }
//        countDownLatch.countDown();
    }

    public int getStartIndex() {
        return startIndex;
    }

    public void setStartIndex(int startIndex) {
        this.startIndex = startIndex;
    }

    public int getEndIndex() {
        return endIndex;
    }

    public void setEndIndex(int endIndex) {
        this.endIndex = endIndex;
    }

    public String[] getDatas() {
        return datas;
    }

    public void setDatas(String[] datas) {
        this.datas = datas;
    }
}

RecursiveTask

執(zhí)行沒有返回結(jié)果的任務(wù)

• 新建 TestRecursiveTask 對任務(wù)進行分割的主線程類

  package com.fengxg.test.recursive.task;
    import com.fengxg.test.recursive.dto.TaskResultDTO;
  
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.RecursiveTask;
  
    /**
     * @Desccription 返回有結(jié)果任務(wù)
     * @auther Fengxg
     * @create 2018/7/25
     */
    public class TestRecursiveTask extends RecursiveTask<TaskResultDTO> {
  
  
    private static final int ONE_TASK_DEAL_NUMBER = 10;
    private List<TaskWorkTask> taskWorkTasks = new ArrayList<TaskWorkTask>();
    /**
     * 計數(shù)器胚鸯，計算器執(zhí)行完骨稿，表示最后一個任務(wù)也跑完
     */
    private CountDownLatch countDownLatch = null;
    /**
     * 計數(shù)器，計算器執(zhí)行完姜钳，表示最后一個任務(wù)也跑完
     */
  
    String[] data = {};
  
    public TestRecursiveTask(String[] data) {
        super();
        this.data = data;
    }
  
    @Override
    protected TaskResultDTO compute() {
        //需要拆分成多少個任務(wù)單數(shù)
        int needSplitCount = 1;
        //剩余多少任務(wù)單
        int leftCount = data.length % ONE_TASK_DEAL_NUMBER;
        needSplitCount = (leftCount == 0?data.length/ONE_TASK_DEAL_NUMBER:data.length/ONE_TASK_DEAL_NUMBER+1);
        //初始化計數(shù)器
        countDownLatch = new CountDownLatch(needSplitCount);
  
        //拆分任務(wù)單
        for(int i=0; i<needSplitCount; i++)
        {
            TaskWorkTask workTask = null;
            if(i == needSplitCount-1 && leftCount!=0)
            {
                workTask = new TaskWorkTask(i*ONE_TASK_DEAL_NUMBER, i*ONE_TASK_DEAL_NUMBER+leftCount,countDownLatch, data,i);
            }else {
                workTask = new TaskWorkTask(i*ONE_TASK_DEAL_NUMBER, (i+1)*ONE_TASK_DEAL_NUMBER,countDownLatch, data,i);
            }
  
            taskWorkTasks.add(workTask);
            //提交到線程池隊列中
            workTask.fork();
        }
  
        //整個任務(wù)直接結(jié)果
        TaskResultDTO taskResultDTO = new TaskResultDTO();
  
        for(TaskWorkTask taskWorkTask: taskWorkTasks){
            //遍歷子任務(wù)返回的結(jié)果集坦冠，統(tǒng)計總結(jié)果
            TaskResultDTO taskWorkTaskResultDTO    = taskWorkTask.join();
            taskResultDTO.setErrorCount(taskResultDTO.getErrorCount() + taskWorkTaskResultDTO.getErrorCount());
            taskResultDTO.setSuccessCount(taskResultDTO.getSuccessCount() + taskWorkTaskResultDTO.getSuccessCount());
            taskResultDTO.setTotalCount(taskResultDTO.getTotalCount() + taskWorkTaskResultDTO.getTotalCount());
        }
        
        return taskResultDTO;
    }
  
  
    public String[] getData() {
        return data;
    }
  
    public void setData(String[] data) {
        this.data = data;
    }
    }
  

• 新建 TaskWorkTask 分片后執(zhí)行具體業(yè)務(wù)任務(wù)的類

package com.fengxg.test.recursive.task;

import com.fengxg.test.recursive.dto.TaskResultDTO;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.RecursiveTask;

/**
* @Desccription 分片后執(zhí)行業(yè)務(wù)任務(wù)的具體類
* @auther Fengxg
* @create 2018/7/25
*/
public class TaskWorkTask extends RecursiveTask<TaskResultDTO>{

  public TaskWorkTask(int startIndex, int endIndex, CountDownLatch countDownLatch, String[] datas, int workNum) {
      super();
      this.startIndex = startIndex;
      this.endIndex = endIndex;
      this.datas = datas;
      this.workNum = workNum;
      this.countDownLatch = countDownLatch;
  }

  private int workNum;
  private int startIndex;
  private int endIndex;
  private CountDownLatch countDownLatch;
  private String[] datas;

  @Override
  protected TaskResultDTO compute() {
      TaskResultDTO taskResultDTO = new TaskResultDTO();
      Long errorCount = 0L;
      Long successCount = 0L;
      Long totalCount = 0L;
      taskResultDTO.setErrorCount(errorCount);
      taskResultDTO.setSuccessCount(successCount);
      taskResultDTO.setTotalCount(totalCount);
      Thread t = Thread.currentThread();
      for (int i = startIndex; i < endIndex; i++) {
          try {
              Thread.sleep(100L);
          } catch (InterruptedException e) {
              e.printStackTrace();
          }
          System.out.println("workNum" + workNum  +"，第" + datas[i] + "任務(wù)," + "執(zhí)行線程:" + t.getName());
          
          if (0 == i%3) {
              taskResultDTO.setErrorCount(taskResultDTO.getErrorCount() + 1);
          } else {
              taskResultDTO.setSuccessCount(taskResultDTO.getSuccessCount() + 1);
          }

          taskResultDTO.setTotalCount(taskResultDTO.getTotalCount() + 1);
      }
      countDownLatch.countDown();
      return taskResultDTO;
  }

  public int getStartIndex() {
      return startIndex;
  }

  public void setStartIndex(int startIndex) {
      this.startIndex = startIndex;
  }

  public int getEndIndex() {
      return endIndex;
  }

  public void setEndIndex(int endIndex) {
      this.endIndex = endIndex;
  }

  public String[] getDatas() {
      return datas;
  }

  public void setDatas(String[] datas) {
      this.datas = datas;
  }
}

調(diào)用 ,測試類

package com.fengxg.test;

import com.fengxg.test.recursive.action.TestRecursiveAction;
import com.fengxg.test.recursive.dto.TaskResultDTO;
import com.fengxg.test.recursive.task.TestRecursiveTask;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;

/**
 * @Desccription
 * @auther Fengxg
 * @create 2018/7/25
 */
public class TestForkJoinMain {
    public static void main(String[] args) throws InterruptedException {
        int count = 50;
        String[] datas= new String[count];
        for (int i = 0; i < count; i++) {
            datas[i] = String.valueOf(i);
        }

        Long start = System.currentTimeMillis();
        /**
         * 有返回值
         */
//        testRecursiveTask(datas);
        /**
         * 無返回值
         */
        testRecursiveAction(datas);
        System.out.println(System.currentTimeMillis() - start);
        Thread.sleep(10000L);
    }

    public static void testRecursiveTask(String[] datas){
        TestRecursiveTask action = new TestRecursiveTask(datas);
        ForkJoinPool fork = new ForkJoinPool();
        Future<TaskResultDTO> future = fork.submit(action);
        try {
           TaskResultDTO taskResultDTO =  future.get();
            System.out.println(taskResultDTO.toString());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }


    /**
     * @author fengxg
     * @description  測試沒有返回結(jié)果的任務(wù)
     * @params []
     * @return void
     * @data 10:15 2018/7/25
     */
    public static void testRecursiveAction(String[] datas){
        TestRecursiveAction action = new TestRecursiveAction(datas);
        ForkJoinPool fork = new ForkJoinPool();
        Future future = fork.submit(action);
        try {
            future.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }

}

So?

? ? 本示例只做fork-join應(yīng)用展示哥桥，后續(xù)會有專門文章對其原理進行探討辙浑。
? ? 對于<font color="red">批量</font>操作<font color="red">獨立任務(wù)</font>的需求，fork/join 架構(gòu)有著明顯優(yōu)勢拟糕，尤其是在任務(wù)量大的時候判呕。 但需要注意的是倦踢，多線程能發(fā)揮優(yōu)勢的環(huán)境是<font color="red">多核CPU(運行環(huán)境)</font>。除此以外侠草，需要使用者 針對需求自行調(diào)試出最佳分片（影響因素主要有CPU核數(shù)辱挥，任務(wù)量規(guī)模）。