Java高效并行执行框架介绍

背景

最近对易企秀推荐系统框架进行了一次升级，将原有详情页的单路召回(基于内容相似性)，调整为可支持基于协同、关联规则、画像标签等多路算法召回模式，召回是从海量的数据里快速拿到相对少量的数据，要快而准，为了满足基本的性能指标，这里的召回模块需要考虑支持并行执行。

某电商架构图参考 易企秀商品推荐框架

Fork/Join框架是JAVA7提供的一个用于并行执行任务的框架，是一个把大任务分割成若干个小任务，最终汇总每个小任务结果后得到大任务结果的框架

框架优势

该框架采用工作窃取算法，比传统的Executor方式更加高效与简单，不需要再单独维护计数器了；工作窃取算法指的是某个线程从其他队列里窃取任务来执行。使用的场景是一个大任务拆分成多个小任务，为了减少线程间的竞争，把这些子任务分别放到不同的队列中，并且每个队列都有单独的线程来执行队列里的任务，线程和队列一一对应。但是会出现这样一种情况：A线程处理完了自己队列的任务，B线程的队列里还有很多任务要处理。A是一个很热情的线程，想过去帮忙，但是如果两个线程访问同一个队列，会产生竞争，所以A想了一个办法，从双端队列的尾部拿任务执行。而B线程永远是从双端队列的头部拿任务执行（任务是一个个独立的小任务），这样感觉A线程像是小偷在窃取B线程的东西一样。

简单示例

package ai.djl.examples.inference;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.RecursiveTask;

public class MyTest extends RecursiveTask<List<Integer>> {
    private int cn = 0;
    public MyTest(int a){
        this.cn = a;
    }
    public static void main(String[] args) {
        long time1 = System.currentTimeMillis();
        MyTest t1 = new MyTest(3);    // 任务1耗时600ms
        MyTest t2 = new MyTest(5);   // 任务2耗时1s
        t1.fork();
        t2.fork();
        List<Integer> l1 = t1.join();
        List<Integer> l2 = t2.join();
        l1.addAll(l2);
        System.out.println(l1+": "+(System.currentTimeMillis() - time1)); //如果串行执行，执行时间会大于1600ms
    }
    @Override
    protected List<Integer> compute() {
        try {
            Thread.sleep(cn*200);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        List<Integer> ls = new ArrayList<>();
        for(int i=0; i<cn; i++){
            ls.add(i);
        }
        return ls;
    }
}