并发编程之Fork/Join框架
一、基本介绍1.1 基本介绍Fork/Join框架是Java 7提供的一个用于并行执行任务的框架 核心思想就是把大任务分割成若干个小任务最终汇总每个小任务结果后得到大任务结果其实现思想与MapReduce有异曲同工之妙。Fork就是把一个大任务切分为若干子任务并行的执行Join就是合并这些子任务的执行结果最后得到这个大任务的结果。比如计算12…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.javapackage 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 IntegerMAX200;static class CalcForJoinTask extends RecursiveTaskInteger{private Integer startValue;//子任务的开始计算的值 private Integer endValue;//子任务结束计算的值 public CalcForJoinTask(Integer startValue, Integer endValue){this.startValuestartValue;this.endValueendValue;}Override protected Integercompute(){//如果当前的数据区间已经小于MAX了那么接下来的计算不需要做拆分 if(endValue-startValueMAX){System.out.println(开始计算startValue:startValue ; endValue:endValue);IntegertotalValue0;for(intithis.startValue;ithis.endValue;i){totalValuei;}returntotalValue;}CalcForJoinTasksubTasknew CalcForJoinTask(startValue,(startValueendValue)/2);subTask.fork();CalcForJoinTaskcalcForJoinTasknew CalcForJoinTask((startValueendValue)/21,endValue);calcForJoinTask.fork();returnsubTask.join()calcForJoinTask.join();}}public static void main(String[]args){CalcForJoinTaskcalcForJoinTasknew CalcForJoinTask(1,10000);ForkJoinPoolpoolnew ForkJoinPool();ForkJoinTaskIntegertaskFuturepool.submit(calcForJoinTask);try{IntegerresulttaskFuture.get();System.out.println(result:result);}catch(InterruptedException e){e.printStackTrace();}catch(ExecutionException e){e.printStackTrace();}}}2.2 业务应用业务背景:对商品信息、商品评论信息、商品销量数据、商家信息进行聚合查询。2.2.1 项目搭建创建maven项目pom.xml?xmlversion1.0encodingUTF-8?projectxmlnshttp://maven.apache.org/POM/4.0.0xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersionparentgroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion2.5.0/versionrelativePath/!-- lookup parent from repository --/parentgroupIdcom.example/groupIdartifactIdspring-boot-fork-join/artifactIdversion0.0.1-SNAPSHOT/versionnamespring-boot-fork-join/namedescriptionDemo projectforSpring Boot/descriptionpropertiesjava.version1.8/java.version/propertiesdependenciesdependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactIdscopetest/scope/dependency/dependenciesbuildpluginsplugingroupIdorg.springframework.boot/groupIdartifactIdspring-boot-maven-plugin/artifactId/plugin/plugins/build/project2.2.2 创建实体类Item.javapackage com.example.entity /** * 商品信息 **/ public class Item{private String productName;private int num;public StringgetProductName(){returnproductName;}public void setProductName(String productName){this.productNameproductName;}public intgetNum(){returnnum;}public void setNum(int num){this.numnum;}}Comment.javapackage com.example.entity /** 商品评论信息 **/ public class Comment{private String name;private String content;public StringgetName(){returnname;}public void setName(String name){this.namename;}public StringgetContent(){returncontent;}public void setContent(String content){this.contentcontent;}}Seller.javapackage com.example.entity /** * 销售信息 **/ public class Seller{private int totalNum;private int sellerNum;public intgetTotalNum(){returntotalNum;}public void setTotalNum(int totalNum){this.totalNumtotalNum;}public intgetSellerNum(){returnsellerNum;}public void setSellerNum(int sellerNum){this.sellerNumsellerNum;}}Shop.javapackage com.example.entity /** * 商家信息 **/ public class Shop{private String name;public StringgetName(){returnname;}public void setName(String name){this.namename;}}Context.javapackage 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.itemitem;}public CommentgetComment(){returncomment;}public void setComment(Comment comment){this.commentcomment;}public SellergetSeller(){returnseller;}public void setSeller(Seller seller){this.sellerseller;}public ShopgetShop(){returnshop;}public void setShop(Shop shop){this.shopshop;}Override public StringtoString(){returnContext{item item , comment comment , seller seller , shop shop };}}2.2.3 创建基础接口和类ILoadDataProcessor.javapackage com.example.basic public interface ILoadDataProcessor{/** * 加载对应的数据 * param context */ void load(Context context);}AbstractLoadDataProcessor.javapackage 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.contextcontext;}}2.2.4 创建服务类CommentService.javapackage com.example.serviceimportorg.springframework.stereotype.Service;/** 评论服务类 **/ Service public class CommentService extends AbstractLoadDataProcessor{Override public void load(Context context){//RPC. Commentcommentnew Comment();comment.setName(XIWANG);comment.setContent(商品质量很好);context.setComment(comment);}}ItemService.javapackage com.example.serviceimportorg.springframework.stereotype.Service;/** 商品服务类 **/ Service public class ItemService extends AbstractLoadDataProcessor{Override public void load(Context context){Itemitemnew Item();item.setNum(100);item.setProductName(键盘);context.setItem(item);}}SellerService.javapackage com.example.serviceimportorg.springframework.stereotype.Service;/** 销量服务类 **/ Service public class SellerService extends AbstractLoadDataProcessor{Override public void load(Context context){Sellersellernew Seller();seller.setSellerNum(100);seller.setTotalNum(1000);context.setSeller(seller);}}ShopService.javapackage com.example.serviceimportorg.springframework.stereotype.Service;/** 商家服务类 **/ Service public class ShopService extends AbstractLoadDataProcessor{Override public void load(Context context){Shopshopnew Shop();shop.setName(令狐冲小店);context.setShop(shop);}}2.2.5 创建聚合任务类ComplexTradeTaskService.javapackage 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 ListAbstractLoadDataProcessortaskDataProcessorsnew 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.applicationContextapplicationContext;taskDataProcessors.add(applicationContext.getBean(SellerService.class));taskDataProcessors.add(applicationContext.getBean(ShopService.class));}}ItemTaskForkJoinDataProcessor.javapackage 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 ListAbstractLoadDataProcessortaskDataProcessorsnew 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.applicationContextapplicationContext;taskDataProcessors.add(applicationContext.getBean(CommentService.class));taskDataProcessors.add(applicationContext.getBean(ItemService.class));taskDataProcessors.add(applicationContext.getBean(ComplexTradeTaskService.class));}}2.2.6 创建控制器类和启动类IndexController.javapackage 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(){Contextcontextnew Context();itemTaskForkJoinDataProcessor.setContext(context);ForkJoinPoolforkJoinPoolnew ForkJoinPool();forkJoinPool.submit(itemTaskForkJoinDataProcessor);returnitemTaskForkJoinDataProcessor.getContext();}}SpringBootForkJoinApplication.javapackage 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);}}2.2.7 测试启动项目后浏览器访问http://localhost:8080/say输出结果:
