[大数据]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
输出结果:

创作打卡挑战赛

赢取流量/现金/CSDN周边激励大奖