Fork/Join框架是Java 7提供的一个用于并行执行任务的框架， 核心思想就是把大任务分割成若干个小任务，最终汇总每个小任务结果后得到大任务结果，其实现思想与MapReduce有异曲同工之妙。
Fork就是把一个大任务切分为若干子任务并行的执行，Join就是合并这些子任务的执行结果，最后得到这个大任务的结果。比如计算1+2+…＋10000，可以分割成10个子任务，每个子任务分别对1000个数进行求和，最终汇总这10个子任务的结果。Fork/Join的运行流程图如下：
Fork/Join框架使用一个巧妙的算法来平衡线程的负载，称为工作窃取（work-stealing）算法。工作窃取的运行流程图如下：
假如我们需要做一个比较大的任务，我们可以把这个任务分割为若干互不依赖的子任务，为了减少线程间的竞争，于是把这些子任务分别放到不同的队列里，并为每个队列创建一个单独的线程来执行队列里的任务，线程和队列一一对应，比如A线程负责处理A队列里的任务。但是有的线程会先把自己队列里的任务干完，而其他线程对应的队列里还有任务等待处理。干完活的线程与其等着，不如去帮其他线程干活，于是它就去其他线程的队列里窃取一个任务来执行。而在这时它们会访问同一个队列，所以为了减少窃取任务线程和被窃取任务线程之间的竞争，通常会使用双端队列，被窃取任务线程永远从双端队列的头部拿任务执行，而窃取任务的线程永远从双端队列的尾部拿任务执行。
工作窃取算法的优点是充分利用线程进行并行计算，并减少了线程间的竞争，其缺点是在某些情况下还是存在竞争，比如双端队列里只有一个任务时。并且消耗了更多的系统资源，比如创建多个线程和多个双端队列。
ForkJoinTask : 基本任务，使用fork、join框架必须创建的对象，提供fork,join操作，常用的三个子类
- RecursiveAction : 无结果返回的任务
- RecursiveTask : 有返回结果的任务
- CountedCompleter：无返回值任务，完成任务后可以触发回调。
ForkJoinTask提供了两个重要的方法：
- fork : 让task异步执行
- join : 让task同步执行，可以获取返回值
ForkJoinPool ： 专门用来运行 ForkJoinTask 的线程池，（在实际使用中，也可以接收
Runnable/Callable任务，但在真正运行时，也会把这些任务封装成 ForkJoinTask类型的任务）
ForkJoinTask 在不显式使用 ForkJoinPool.execute/invoke/submit() 方法进行执行的情况下，也可以使用自己的fork/invoke方法进行执行
ForkJoinExample.java
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;
public class ForkJoinExample {
//java8 parallStream
//针对一个数字，做计算。
private static final Integer MAX=200;
static class CalcForJoinTask extends RecursiveTask {
private Integer startValue; //子任务的开始计算的值
private Integer endValue; //子任务结束计算的值
public CalcForJoinTask(Integer startValue, Integer endValue) {
this.startValue = startValue;
this.endValue = endValue;
}
@Override
protected Integer compute() {
//如果当前的数据区间已经小于MAX了，那么接下来的计算不需要做拆分
if(endValue-startValue
System.out.println("开始计算：startValue:"+startValue+" ; endValue:"+endValue);
Integer totalValue=0;
for(int i=this.startValue;i<=this.endValue;i++){
totalValue+=i;
}
return totalValue;
}
CalcForJoinTask subTask=new CalcForJoinTask(startValue,(startValue+endValue)/2);
subTask.fork();
CalcForJoinTask calcForJoinTask=new CalcForJoinTask((startValue+endValue)/2+1,endValue);
calcForJoinTask.fork();
return subTask.join()+calcForJoinTask.join();
}
}
public static void main(String[] args) {
CalcForJoinTask calcForJoinTask=new CalcForJoinTask(1,10000);
ForkJoinPool pool=new ForkJoinPool();
ForkJoinTask taskFuture=pool.submit(calcForJoinTask);
try {
Integer result=taskFuture.get();
System.out.println("result:"+result);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
}
2.2 业务应用
业务背景:对商品信息、商品评论信息、商品销量数据、商家信息进行聚合查询。
2.2.1 项目搭建创建maven项目
pom.xml
2.2.2 创建实体类4.0.0 org.springframework.boot spring-boot-starter-parent 2.5.0 com.example spring-boot-fork-join 0.0.1-SNAPSHOT spring-boot-fork-join Demo project for Spring Boot 1.8 org.springframework.boot spring-boot-starter-web org.springframework.boot spring-boot-starter-test test org.springframework.boot spring-boot-maven-plugin
Item.java
public class Item {
private String productName;
private int num;
public String getProductName() {
return productName;
}
public void setProductName(String productName) {
this.productName = productName;
}
public int getNum() {
return num;
}
public void setNum(int num) {
this.num = num;
}
}
Comment.java
public class Comment {
private String name;
private String content;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getContent() {
return content;
}
public void setContent(String content) {
this.content = content;
}
}
Seller.java
public class Seller {
private int totalNum;
private int sellerNum;
public int getTotalNum() {
return totalNum;
}
public void setTotalNum(int totalNum) {
this.totalNum = totalNum;
}
public int getSellerNum() {
return sellerNum;
}
public void setSellerNum(int sellerNum) {
this.sellerNum = sellerNum;
}
}
Shop.java
public class Shop {
private String name;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
}
Context.java
public class Context {
private Item item; //商品
private Comment comment; //评论
private Seller seller; //销售信息
private Shop shop; //店铺信息
public Item getItem() {
return item;
}
public void setItem(Item item) {
this.item = item;
}
public Comment getComment() {
return comment;
}
public void setComment(Comment comment) {
this.comment = comment;
}
public Seller getSeller() {
return seller;
}
public void setSeller(Seller seller) {
this.seller = seller;
}
public Shop getShop() {
return shop;
}
public void setShop(Shop shop) {
this.shop = shop;
}
@Override
public String toString() {
return "Context{" +
"item=" + item +
", comment=" + comment +
", seller=" + seller +
", shop=" + shop +
'}';
}
}
