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

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<Integer>{
        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<MAX){
                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<Integer>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

<?xml version="1.0"encoding="UTF-8"?><project xmlns="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 project for Spring 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

/**
 * 商品信息
 **/
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 +
                '}';}}

2.2.3 创建基础接口和类

ILoadDataProcessor.java

public interface ILoadDataProcessor {

    /**
     * 加载对应的数据
     * @param context
     */
    void load(Context context);}

AbstractLoadDataProcessor.java

import java.util.concurrent.RecursiveAction;

/**
数据加载抽象类
 **/
public abstract class AbstractLoadDataProcessor extends RecursiveAction implements ILoadDataProcessor{

    protected Context context;

    @Override
    protected void compute(){
        load(context); //调用子类的具体实现
    }

    public Context getContext(){
        this.join(); //得到一个聚合的结果
        return context;}

    public void setContext(Context context){
        this.context = context;}}

2.2.4 创建服务类

CommentService.java

import org.springframework.stereotype.Service;

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

ItemService.java

import org.springframework.stereotype.Service;

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

SellerService.java

import org.springframework.stereotype.Service;

/**
销量服务类
 **/
@Service
public class SellerService extends AbstractLoadDataProcessor{

    @Override
   public void load(Context context){
        Seller seller=new Seller();
        seller.setSellerNum(100);
        seller.setTotalNum(1000);
        context.setSeller(seller);}}

ShopService.java

import org.springframework.stereotype.Service;

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

2.2.5 创建聚合任务类

ComplexTradeTaskService.java

import org.springframework.beans.BeansException;import org.springframework.context.ApplicationContext;import org.springframework.context.ApplicationContextAware;import org.springframework.stereotype.Service;import java.util.ArrayList;import java.util.List;import java.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 Context getContext(){
        this.taskDataProcessors.forEach(ForkJoinTask::join);return super.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

import org.springframework.beans.BeansException;import org.springframework.context.ApplicationContext;import org.springframework.context.ApplicationContextAware;import org.springframework.stereotype.Service;import java.util.ArrayList;import java.util.List;import java.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 Context getContext(){
        //ForkJoinTask::join  java8方法引用
        // * 构造引用
        // * 静态方法引用
        // * 实例方法引用
        this.taskDataProcessors.forEach(ForkJoinTask::join);return super.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

import org.springframework.beans.factory.annotation.Autowired;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.RestController;import java.util.concurrent.ForkJoinPool;

@RestController
public class IndexController {

    @Autowired
    ItemTaskForkJoinDataProcessor itemTaskForkJoinDataProcessor;

    @GetMapping("/say")
    public Context index(){
        Context context=new Context();
        itemTaskForkJoinDataProcessor.setContext(context);
        ForkJoinPool forkJoinPool=new ForkJoinPool();
        forkJoinPool.submit(itemTaskForkJoinDataProcessor);return itemTaskForkJoinDataProcessor.getContext();}}

SpringBootForkJoinApplication.java

import org.springframework.boot.SpringApplication;import org.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
输出结果: