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