并发编程之Fork/Join框架-开发者社区

一、基本介绍

1.1 基本介绍

Fork/Join框架是Java 7提供的一个用于并行执行任务的框架，核心思想就是把大任务分割成若干个小任务，最终汇总每个小任务结果后得到大任务结果，其实现思想与MapReduce有异曲同工之妙。

Fork就是把一个大任务切分为若干子任务并行的执行，Join就是合并这些子任务的执行结果，最后得到这个大任务的结果。比如计算1+2+…＋10000，可以分割成10个子任务，每个子任务分别对1000个数进行求和，最终汇总这10个子任务的结果。Fork/Join的运行流程图如下：

Fork/Join框架使用一个巧妙的算法来平衡线程的负载，称为工作窃取（work-stealing）算法。工作窃取的运行流程图如下：

假如我们需要做一个比较大的任务，我们可以把这个任务分割为若干互不依赖的子任务，为了减少线程间的竞争，于是把这些子任务分别放到不同的队列里，并为每个队列创建一个单独的线程来执行队列里的任务，线程和队列一一对应，比如A线程负责处理A队列里的任务。但是有的线程会先把自己队列里的任务干完，而其他线程对应的队列里还有任务等待处理。干完活的线程与其等着，不如去帮其他线程干活，于是它就去其他线程的队列里窃取一个任务来执行。而在这时它们会访问同一个队列，所以为了减少窃取任务线程和被窃取任务线程之间的竞争，通常会使用双端队列，被窃取任务线程永远从双端队列的头部拿任务执行，而窃取任务的线程永远从双端队列的尾部拿任务执行。
工作窃取算法的优点是充分利用线程进行并行计算，并减少了线程间的竞争，其缺点是在某些情况下还是存在竞争，比如双端队列里只有一个任务时。并且消耗了更多的系统资源，比如创建多个线程和多个双端队列。

1.2 API代码分析

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方法进行执行

二、应用

2.1 简单应用

ForkJoinExample.java

package cn.zxj.forkjoin;importjava.util.concurrent.ExecutionException;importjava.util.concurrent.ForkJoinPool;importjava.util.concurrent.ForkJoinTask;importjava.util.concurrent.RecursiveTask;public class ForkJoinExample{//java8 parallStream //针对一个数字，做计算。 private static final IntegerMAX=200;static class CalcForJoinTask extends RecursiveTask<Integer>{private Integer startValue;//子任务的开始计算的值 private Integer endValue;//子任务结束计算的值 public CalcForJoinTask(Integer startValue, Integer endValue){this.startValue=startValue;this.endValue=endValue;}@Override protected Integercompute(){//如果当前的数据区间已经小于MAX了，那么接下来的计算不需要做拆分 if(endValue-startValue<MAX){System.out.println("开始计算：startValue:"+startValue+" ; endValue:"+endValue);IntegertotalValue=0;for(inti=this.startValue;i<=this.endValue;i++){totalValue+=i;}returntotalValue;}CalcForJoinTasksubTask=new CalcForJoinTask(startValue,(startValue+endValue)/2);subTask.fork();CalcForJoinTaskcalcForJoinTask=new CalcForJoinTask((startValue+endValue)/2+1,endValue);calcForJoinTask.fork();returnsubTask.join()+calcForJoinTask.join();}}public static void main(String[]args){CalcForJoinTaskcalcForJoinTask=new CalcForJoinTask(1,10000);ForkJoinPoolpool=new ForkJoinPool();ForkJoinTask<Integer>taskFuture=pool.submit(calcForJoinTask);try{Integerresult=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

<?xmlversion="1.0"encoding="UTF-8"?><projectxmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.5.0</version><relativePath/><!-- lookup parent from repository --></parent><groupId>com.example</groupId><artifactId>spring-boot-fork-join</artifactId><version>0.0.1-SNAPSHOT</version><name>spring-boot-fork-join</name><description>Demo projectforSpring Boot</description><properties><java.version>1.8</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

2.2.2 创建实体类

Item.java

package com.example.entity /** * 商品信息 **/ public class Item{private String productName;private int num;public StringgetProductName(){returnproductName;}public void setProductName(String productName){this.productName=productName;}public intgetNum(){returnnum;}public void setNum(int num){this.num=num;}}

Comment.java

package com.example.entity /** 商品评论信息 **/ public class Comment{private String name;private String content;public StringgetName(){returnname;}public void setName(String name){this.name=name;}public StringgetContent(){returncontent;}public void setContent(String content){this.content=content;}}

Seller.java

package com.example.entity /** * 销售信息 **/ public class Seller{private int totalNum;private int sellerNum;public intgetTotalNum(){returntotalNum;}public void setTotalNum(int totalNum){this.totalNum=totalNum;}public intgetSellerNum(){returnsellerNum;}public void setSellerNum(int sellerNum){this.sellerNum=sellerNum;}}

Shop.java

package com.example.entity /** * 商家信息 **/ public class Shop{private String name;public StringgetName(){returnname;}public void setName(String name){this.name=name;}}

Context.java

package com.example.entity /** * 聚合信息 **/ public class Context{private Item item;//商品 private Comment comment;//评论 private Seller seller;//销售信息 private Shop shop;//店铺信息 public ItemgetItem(){returnitem;}public void setItem(Item item){this.item=item;}public CommentgetComment(){returncomment;}public void setComment(Comment comment){this.comment=comment;}public SellergetSeller(){returnseller;}public void setSeller(Seller seller){this.seller=seller;}public ShopgetShop(){returnshop;}public void setShop(Shop shop){this.shop=shop;}@Override public StringtoString(){return"Context{"+"item="+ item +", comment="+ comment +", seller="+ seller +", shop="+ shop +'}';}}

2.2.3 创建基础接口和类

ILoadDataProcessor.java

package com.example.basic public interface ILoadDataProcessor{/** * 加载对应的数据 * @param context */ void load(Context context);}

AbstractLoadDataProcessor.java

package com.example.basicimportjava.util.concurrent.RecursiveAction;/** 数据加载抽象类 **/ public abstract class AbstractLoadDataProcessor extends RecursiveAction implements ILoadDataProcessor{protected Context context;@Override protected voidcompute(){load(context);//调用子类的具体实现}public ContextgetContext(){this.join();//得到一个聚合的结果returncontext;}public void setContext(Context context){this.context=context;}}

2.2.4 创建服务类

CommentService.java

package com.example.serviceimportorg.springframework.stereotype.Service;/** 评论服务类 **/ @Service public class CommentService extends AbstractLoadDataProcessor{@Override public void load(Context context){//RPC. Commentcomment=new Comment();comment.setName("XIWANG");comment.setContent("商品质量很好");context.setComment(comment);}}

ItemService.java

package com.example.serviceimportorg.springframework.stereotype.Service;/** 商品服务类 **/ @Service public class ItemService extends AbstractLoadDataProcessor{@Override public void load(Context context){Itemitem=new Item();item.setNum(100);item.setProductName("键盘");context.setItem(item);}}

SellerService.java

package com.example.serviceimportorg.springframework.stereotype.Service;/** 销量服务类 **/ @Service public class SellerService extends AbstractLoadDataProcessor{@Override public void load(Context context){Sellerseller=new Seller();seller.setSellerNum(100);seller.setTotalNum(1000);context.setSeller(seller);}}

ShopService.java

package com.example.serviceimportorg.springframework.stereotype.Service;/** 商家服务类 **/ @Service public class ShopService extends AbstractLoadDataProcessor{@Override public void load(Context context){Shopshop=new Shop();shop.setName("令狐冲小店");context.setShop(shop);}}

2.2.5 创建聚合任务类

ComplexTradeTaskService.java

package com.example.serviceimportorg.springframework.beans.BeansException;importorg.springframework.context.ApplicationContext;importorg.springframework.context.ApplicationContextAware;importorg.springframework.stereotype.Service;importjava.util.ArrayList;importjava.util.List;importjava.util.concurrent.ForkJoinTask;/** 销量和商家聚合任务类 **/ @Service public class ComplexTradeTaskService extends AbstractLoadDataProcessor implements ApplicationContextAware{ApplicationContext applicationContext;private List<AbstractLoadDataProcessor>taskDataProcessors=new ArrayList<>();@Override public void load(Context context){taskDataProcessors.forEach(abstractLoadDataProcessor->{abstractLoadDataProcessor.setContext(this.context);abstractLoadDataProcessor.fork();//创建一个fork task});}@Override public ContextgetContext(){this.taskDataProcessors.forEach(ForkJoinTask::join);returnsuper.getContext();}@Override public void setApplicationContext(ApplicationContext applicationContext)throws BeansException{this.applicationContext=applicationContext;taskDataProcessors.add(applicationContext.getBean(SellerService.class));taskDataProcessors.add(applicationContext.getBean(ShopService.class));}}

ItemTaskForkJoinDataProcessor.java

package com.example.serviceimportorg.springframework.beans.BeansException;importorg.springframework.context.ApplicationContext;importorg.springframework.context.ApplicationContextAware;importorg.springframework.stereotype.Service;importjava.util.ArrayList;importjava.util.List;importjava.util.concurrent.ForkJoinTask;/** * 全局聚合任务 **/ @Service public class ItemTaskForkJoinDataProcessor extends AbstractLoadDataProcessor implements ApplicationContextAware{ApplicationContext applicationContext;private List<AbstractLoadDataProcessor>taskDataProcessors=new ArrayList<>();@Override public void load(Context context){taskDataProcessors.forEach(abstractLoadDataProcessor->{abstractLoadDataProcessor.setContext(this.context);abstractLoadDataProcessor.fork();//创建一个fork task});}@Override public ContextgetContext(){//ForkJoinTask::join java8方法引用 // * 构造引用 // * 静态方法引用 // * 实例方法引用 this.taskDataProcessors.forEach(ForkJoinTask::join);returnsuper.getContext();}@Override public void setApplicationContext(ApplicationContext applicationContext)throws BeansException{this.applicationContext=applicationContext;taskDataProcessors.add(applicationContext.getBean(CommentService.class));taskDataProcessors.add(applicationContext.getBean(ItemService.class));taskDataProcessors.add(applicationContext.getBean(ComplexTradeTaskService.class));}}

2.2.6 创建控制器类和启动类

IndexController.java

package com.example.controllerimportorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.web.bind.annotation.GetMapping;importorg.springframework.web.bind.annotation.RestController;importjava.util.concurrent.ForkJoinPool;@RestController public class IndexController{@Autowired ItemTaskForkJoinDataProcessor itemTaskForkJoinDataProcessor;@GetMapping("/say")public Contextindex(){Contextcontext=new Context();itemTaskForkJoinDataProcessor.setContext(context);ForkJoinPoolforkJoinPool=new ForkJoinPool();forkJoinPool.submit(itemTaskForkJoinDataProcessor);returnitemTaskForkJoinDataProcessor.getContext();}}

SpringBootForkJoinApplication.java

package com.exampleimportorg.springframework.boot.SpringApplication;importorg.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication public class SpringBootForkJoinApplication{public static void main(String[]args){SpringApplication.run(SpringBootForkJoinApplication.class, args);}}