简体字、冯|java11出来了，java8里边最好用的并发框架 fork-join怎么玩，会了吗？

title: 2018-8-3 聊聊 fork-join框架
tags: java，并发，线程池
grammar_cjkRuby: true

what？

Fork/Join框架是Java7提供了的一个用于并行执行任务的框架，是一个把大任务分割成若干个小任务，最终汇总每个小任务结果后得到大任务结果的框架。

why?

同样是线程池，为什么我们要用Fork/Join不用别的。

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

先执行完当前线程的任务，直接去从其他为执行完任务的线程中窃取任务执行；

图：

任务窃取

如图线程1先执行完自己的4个任务，线程2还有没执行的任务，线程1接着从线程2中获取未执行的任务。实现是基于双向队列

How？

RecursiveAction

执行没有返回结果的任务

新建TestRecursiveAction 对任务进行分割的主线程类

package com.fengxg.test.recursive.action;

import java.util.concurrent.RecursiveAction;

/**
 * @Desccription 用于没有返回结果的任务，对任务进行分割的主线程
 * @auther Fengxg
 * @create 2018/7/25
 */
public class TestRecursiveAction extends RecursiveAction {

    private static final int ONE_TASK_DEAL_NUMBER = 10;

    /**
     * 计数器，计算器执行完，表示最后一个任务也跑完
     */

    String[] data = {};

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

    @Override
    protected void compute() {
        //需要拆分成多少个任务单数
        int needSplitCount = 1;
        //剩余多少任务单
        int leftCount = data.length % ONE_TASK_DEAL_NUMBER;
        needSplitCount = (leftCount == 0?data.length/ONE_TASK_DEAL_NUMBER:data.length/ONE_TASK_DEAL_NUMBER+1);
        //初始化计数器
//        countDownLatch = new CountDownLatch(needSplitCount);

        //拆分任务单
        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 分片后执行具体业务任务的类

package com.fengxg.test.recursive.action;

import java.util.concurrent.RecursiveAction;

/**
 * @Desccription 分片后执行业务任务的具体类
 * @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] + "任务," + "执行线程:" + 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

执行没有返回结果的任务

新建 TestRecursiveTask 对任务进行分割的主线程类

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 返回有结果任务
   * @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>();
  /**
   * 计数器，计算器执行完，表示最后一个任务也跑完
   */
  private CountDownLatch countDownLatch = null;
  /**
   * 计数器，计算器执行完，表示最后一个任务也跑完
   */

  String[] data = {};

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

  @Override
  protected TaskResultDTO compute() {
      //需要拆分成多少个任务单数
      int needSplitCount = 1;
      //剩余多少任务单
      int leftCount = data.length % ONE_TASK_DEAL_NUMBER;
      needSplitCount = (leftCount == 0?data.length/ONE_TASK_DEAL_NUMBER:data.length/ONE_TASK_DEAL_NUMBER+1);
      //初始化计数器
      countDownLatch = new CountDownLatch(needSplitCount);

      //拆分任务单
      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();
      }

      //整个任务直接结果
      TaskResultDTO taskResultDTO = new TaskResultDTO();

      for(TaskWorkTask taskWorkTask: taskWorkTasks){
          //遍历子任务返回的结果集，统计总结果
          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 分片后执行具体业务任务的类

package com.fengxg.test.recursive.task;

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

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

/**
* @Desccription 分片后执行业务任务的具体类
* @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] + "任务," + "执行线程:" + 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;
  }
}

调用 ,测试类

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  测试没有返回结果的任务
     * @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应用展示，后续会有专门文章对其原理进行探讨。
对于批量操作独立任务的需求，fork/join 架构有着明显优势，尤其是在任务量大的时候。但需要注意的是，多线程能发挥优势的环境是多核CPU(运行环境)。除此以外，需要使用者针对需求自行调试出最佳分片（影响因素主要有CPU核数，任务量规模）。