后端思维篇：手把手教你写一个并行调用模板

前言

大家好，我是捡田螺的小男孩。

本文是后端思维专栏的第二篇哈。上一篇36个设计接口的锦囊，得到非常多小伙伴的认可。 36个设计接口的锦囊中也提到一个点：就是使用并行调用优化接口。所以接下来就快马加鞭，写第二篇：手把手教你写一个并行调用模板。

一个串行调用的例子（App首页信息查询）

CompletionService实现并行调用

抽取通用的并行调用方法

代码思考以及设计模式应用

思考总结

公众号：捡田螺的小男孩

1. 一个串行调用的例子

如果让你设计一个APP首页查询的接口，它需要查用户信息、需要查banner信息、需要查标签信息等等。一般情况，小伙伴会实现如下：

public AppHeadInfoResponse queryAppHeadInfo(AppInfoReq req) {    //查用户信息    UserInfoParam userInfoParam = buildUserParam(req);    UserInfoDTO userInfoDTO = userService.queryUserInfo(userInfoParam);    //查banner信息    BannerParam bannerParam = buildBannerParam(req);    BannerDTO bannerDTO = bannerService.queryBannerInfo(bannerParam);    //查标签信息    LabelParam labelParam = buildLabelParam(req);    LabelDTO labelDTO = labelService.queryLabelInfo(labelParam);    //组装结果    return buildResponse(userInfoDTO,bannerDTO,labelDTO);}

这段代码会有什么问题嘛？ 其实这是一段挺正常的代码，但是这个方法实现中，查询用户、banner、标签信息，是串行的，如果查询用户信息200ms，查询banner信息100ms，查询标签信息200ms的话，耗时就是500ms啦。

其实为了优化性能，我们可以修改为并行调用的方式，耗时可以降为200ms，如下图所示：

2. CompletionService实现并行调用

对于上面的例子，如何实现并行调用呢？

有小伙伴说，可以使用Future+Callable实现多个任务的并行调用。但是线程池执行批量任务时，返回值用Future的get()获取是阻塞的，如果前一个任务执行比较耗时的话，get（）方法会阻塞，形成排队等待的情况。

而CompletionService是对定义ExecutorService进行了包装，可以一边生成任务,一边获取任务的返回值。让这两件事分开执行,任务之间不会互相阻塞，可以获取最先完成的任务结果。

CompletionService的实现原理比较简单，底层通过FutureTask+阻塞队列，实现了任务先完成的话，可优先获取到。也就是说任务执行结果按照完成的先后顺序来排序，先完成可以优化获取到。内部有一个先进先出的阻塞队列，用于保存已经执行完成的Future，你调用CompletionService的poll或take方法即可获取到一个已经执行完成的Future，进而通过调用Future接口实现类的get方法获取最终的结果。

接下来，我们来看下，如何用CompletionService，实现并行查询APP首页信息哈。思考步骤如下：

我们先把查询用户信息的任务，放到线程池，如下：

ExecutorService executor = Executors.newFixedThreadPool(10);//查询用户信息CompletionService<UserInfoDTO> userDTOCompletionService = new ExecutorCompletionService<UserInfoDTO>(executor);Callable<UserInfoDTO> userInfoDTOCallableTask = () -> {   UserInfoParam userInfoParam = buildUserParam(req);   return userService.queryUserInfo(userInfoParam);};userDTOCompletionService.submit(userInfoDTOCallableTask);

但是如果想把查询banner信息的任务，也放到这个线程池的话，发现不好放了，因为返回类型不一样，一个是UserInfoDTO，另外一个是BannerDTO。那这时候，我们是不是把泛型声明为Object即可，因为所有对象都是继承于Object的？如下：

ExecutorService executor = Executors.newFixedThreadPool(10);//查询用户信息CompletionService<Object> baseDTOCompletionService = new ExecutorCompletionService<Object>(executor);Callable<Object> userInfoDTOCallableTask = () -> {    UserInfoParam userInfoParam = buildUserParam(req);    return userService.queryUserInfo(userInfoParam);};//banner信息任务Callable<Object> bannerDTOCallableTask = () -> {    BannerParam bannerParam = buildBannerParam(req);    return bannerService.queryBannerInfo(bannerParam);};//提交用户信息任务baseDTOCompletionService.submit(userInfoDTOCallableTask);//提交banner信息任务baseDTOCompletionService.submit(bannerDTOCallableTask);

这里会有个问题，就是获取返回值的时候，我们不知道哪个Object是用户信息的DTO，哪个是BannerDTO？怎么办呢？这时候，我们可以在参数里面做个扩展嘛，即参数声明为一个基础对象BaseRspDTO，再搞个泛型放Object数据的，然后基础对象BaseRspDTO有个区分是UserDTO还是BannerDTO的唯一标记属性key。代码如下：

public class BaseRspDTO<T extends Object> {    //区分是DTO返回的唯一标记，比如是UserInfoDTO还是BannerDTO    private String key;    //返回的data    private T data;    public String getKey() {        return key;    }    public void setKey(String key) {        this.key = key;    }    public T getData() {        return data;    }    public void setData(T data) {        this.data = data;    }}//并行查询App首页信息public AppHeadInfoResponse parallelQueryAppHeadPageInfo(AppInfoReq req) {    long beginTime = System.currentTimeMillis();    System.out.println("开始并行查询app首页信息，开始时间：" + beginTime);    ExecutorService executor = Executors.newFixedThreadPool(10);    CompletionService<BaseRspDTO<Object>> baseDTOCompletionService = new ExecutorCompletionService<BaseRspDTO<Object>>(executor);    //查询用户信息任务    Callable<BaseRspDTO<Object>> userInfoDTOCallableTask = () -> {        UserInfoParam userInfoParam = buildUserParam(req);        UserInfoDTO userInfoDTO = userService.queryUserInfo(userInfoParam);        BaseRspDTO<Object> userBaseRspDTO = new BaseRspDTO<Object>();        userBaseRspDTO.setKey("userInfoDTO");        userBaseRspDTO.setData(userInfoDTO);        return userBaseRspDTO;    };    //banner信息查询任务    Callable<BaseRspDTO<Object>> bannerDTOCallableTask = () -> {        BannerParam bannerParam = buildBannerParam(req);        BannerDTO bannerDTO = bannerService.queryBannerInfo(bannerParam);        BaseRspDTO<Object> bannerBaseRspDTO = new BaseRspDTO<Object>();        bannerBaseRspDTO.setKey("bannerDTO");        bannerBaseRspDTO.setData(bannerDTO);        return bannerBaseRspDTO;    };    //label信息查询任务    Callable<BaseRspDTO<Object>> labelDTODTOCallableTask = () -> {        LabelParam labelParam = buildLabelParam(req);        LabelDTO labelDTO = labelService.queryLabelInfo(labelParam);        BaseRspDTO<Object> labelBaseRspDTO = new BaseRspDTO<Object>();        labelBaseRspDTO.setKey("labelDTO");        labelBaseRspDTO.setData(labelDTO);        return labelBaseRspDTO;    };    //提交用户信息任务    baseDTOCompletionService.submit(userInfoDTOCallableTask);    //提交banner信息任务    baseDTOCompletionService.submit(bannerDTOCallableTask);    //提交label信息任务    baseDTOCompletionService.submit(labelDTODTOCallableTask);    UserInfoDTO userInfoDTO = null;    BannerDTO bannerDTO = null;    LabelDTO labelDTO = null;    try {        //因为提交了3个任务，所以获取结果次数是3        for (int i = 0; i < 3; i++) {            Future<BaseRspDTO<Object>> baseRspDTOFuture = baseDTOCompletionService.poll(1, TimeUnit.SECONDS);            BaseRspDTO baseRspDTO = baseRspDTOFuture.get();            if ("userInfoDTO".equals(baseRspDTO.getKey())) {                userInfoDTO = (UserInfoDTO) baseRspDTO.getData();            } else if ("bannerDTO".equals(baseRspDTO.getKey())) {                bannerDTO = (BannerDTO) baseRspDTO.getData();            } else if ("labelDTO".equals(baseRspDTO.getKey())) {                labelDTO = (LabelDTO) baseRspDTO.getData();            }        }    } catch (InterruptedException e) {        e.printStackTrace();    } catch (ExecutionException e) {        e.printStackTrace();    }    System.out.println("结束并行查询app首页信息,总耗时：" + (System.currentTimeMillis() - beginTime));    return buildResponse(userInfoDTO, bannerDTO, labelDTO);}

到这里为止，一个基于CompletionService实现并行调用的例子已经实现啦。是不是很开心，哈哈。

3. 抽取通用的并行调用方法

我们回过来观察下第2小节，查询app首页信息的demo：CompletionService实现了并行调用。大家有没有什么其他想法呢？比如,假设别的业务场景，也想通过并行调用优化，那是不是也得搞一套类似第2小节的代码。所以，我们是不是可以抽取一个通用的并行方法，让别的场景也可以用，对吧？这就是后端思维啦！

基于第2小节的代码，我们如何抽取通用并行调用方法呢。

首先，这个通用的并行调用方法，不能跟业务相关的属性挂钩，对吧，所以方法的入参应该有哪些呢？

方法的入参，可以有Callable对吧。因为并行，肯定是多个Callable任务的。所以，入参应该是一个Callable的数组。再然后，基于上面的APP首页查询的例子，Callable里面得带BaseRspDTO泛型，对吧？因此入参就是List<Callable<BaseRspDTO<Object>>> list。

那并行调用的出参呢？ 你有多个Callable的任务，是不是得有多个对应的返回，因此，你的出参可以是List<BaseRspDTO<Object>>。我们抽取的通用并行调用模板，就可以写成酱紫：

    public List<BaseRspDTO<Object>> executeTask(List<Callable<BaseRspDTO<Object>>> taskList) {        List<BaseRspDTO<Object>> resultList = new ArrayList<>();        //校验参数        if (taskList == null || taskList.size() == 0) {            return resultList;        }        ExecutorService executor = Executors.newFixedThreadPool(10);        CompletionService<BaseRspDTO<Object>> baseDTOCompletionService = new ExecutorCompletionService<BaseRspDTO<Object>>(executor);        //提交任务        for (Callable<BaseRspDTO<Object>> task : taskList) {            baseDTOCompletionService.submit(task);        }        try {            //遍历获取结果            for (int i = 0; i < taskList.size(); i++) {                Future<BaseRspDTO<Object>> baseRspDTOFuture = baseDTOCompletionService.poll(2, TimeUnit.SECONDS);                resultList.add(baseRspDTOFuture.get());            }        } catch (InterruptedException e) {            e.printStackTrace();        } catch (ExecutionException e) {            e.printStackTrace();        }        return resultList;    }

既然我们是抽取通用的并行调用方法，那以上的方法是否还有哪些地方需要改进的呢？

第一个可以优化的地方，就是executor线程池，比如有些业务场景想用A线程池，有些业务想用B线程池，那么，这个方法，就不通用啦，对吧。我们可以把线程池以参数的实行提供出来，给调用方自己控制。

第二个可以优化的地方，就是CompletionService的poll方法获取时，超时时间是写死的。因为不同业务场景，超时时间可能不一样。所以，超时时间也是可以以参数形式放出来，给调用方自己控制。

我们再次优化一下这个通用的并行调用模板，代码如下：

public List<BaseRspDTO<Object>> executeTask(List<Callable<BaseRspDTO<Object>>> taskList, long timeOut, ExecutorService executor) {    List<BaseRspDTO<Object>> resultList = new ArrayList<>();    //校验参数    if (taskList == null || taskList.size() == 0) {        return resultList;    }    if (executor == null) {        return resultList;    }    if (timeOut <= 0) {        return resultList;     }    //提交任务    CompletionService<BaseRspDTO<Object>> baseDTOCompletionService = new ExecutorCompletionService<BaseRspDTO<Object>>(executor);    for (Callable<BaseRspDTO<Object>> task : taskList) {        baseDTOCompletionService.submit(task);    }    try {        //遍历获取结果        for (int i = 0; i < taskList.size(); i++) {          Future<BaseRspDTO<Object>> baseRspDTOFuture = baseDTOCompletionService.poll(timeOut, TimeUnit.SECONDS);          resultList.add(baseRspDTOFuture.get());        }      } catch (InterruptedException e) {        e.printStackTrace();    } catch (ExecutionException e) {        e.printStackTrace();    }    return resultList;}

以后别的场景也需要用到并行调用的话，直接调用你的这个方法即可，是不是有点小小的成就感啦，哈哈。

4. 代码思考以及设计模式应用

我们把抽取的那个公用的并行调用方法，应用到App首页信息查询的例子，代码如下：

public AppHeadInfoResponse parallelQueryAppHeadPageInfo1(AppInfoReq req) {        long beginTime = System.currentTimeMillis();        System.out.println("开始并行查询app首页信息，开始时间：" + beginTime);        //用户信息查询任务        Callable<BaseRspDTO<Object>> userInfoDTOCallableTask = () -> {            UserInfoParam userInfoParam = buildUserParam(req);            UserInfoDTO userInfoDTO = userService.queryUserInfo(userInfoParam);            BaseRspDTO<Object> userBaseRspDTO = new BaseRspDTO<Object>();            userBaseRspDTO.setKey("userInfoDTO");            userBaseRspDTO.setData(userInfoDTO);            return userBaseRspDTO;        };        //banner信息查询任务        Callable<BaseRspDTO<Object>> bannerDTOCallableTask = () -> {            BannerParam bannerParam = buildBannerParam(req);            BannerDTO bannerDTO = bannerService.queryBannerInfo(bannerParam);            BaseRspDTO<Object> bannerBaseRspDTO = new BaseRspDTO<Object>();            bannerBaseRspDTO.setKey("bannerDTO");            bannerBaseRspDTO.setData(bannerDTO);            return bannerBaseRspDTO;        };        //label信息查询任务        Callable<BaseRspDTO<Object>> labelDTODTOCallableTask = () -> {            LabelParam labelParam = buildLabelParam(req);            LabelDTO labelDTO = labelService.queryLabelInfo(labelParam);            BaseRspDTO<Object> labelBaseRspDTO = new BaseRspDTO<Object>();            labelBaseRspDTO.setKey("labelDTO");            labelBaseRspDTO.setData(labelDTO);            return labelBaseRspDTO;        };        List<Callable<BaseRspDTO<Object>>> taskList = new ArrayList<>();        taskList.add(userInfoDTOCallableTask);        taskList.add(bannerDTOCallableTask);        taskList.add(labelDTODTOCallableTask);        ExecutorService executor = Executors.newFixedThreadPool(10);        List<BaseRspDTO<Object>> resultList = parallelInvokeCommonService.executeTask(taskList, 3, executor);        if (resultList == null || resultList.size() == 0) {            return new AppHeadInfoResponse();        }        UserInfoDTO userInfoDTO = null;        BannerDTO bannerDTO = null;        LabelDTO labelDTO = null;        //遍历结果        for (int i = 0; i < resultList.size(); i++) {            BaseRspDTO baseRspDTO = resultList.get(i);            if ("userInfoDTO".equals(baseRspDTO.getKey())) {                userInfoDTO = (UserInfoDTO) baseRspDTO.getData();            } else if ("bannerDTO".equals(baseRspDTO.getKey())) {                bannerDTO = (BannerDTO) baseRspDTO.getData();            } else if ("labelDTO".equals(baseRspDTO.getKey())) {                labelDTO = (LabelDTO) baseRspDTO.getData();            }        }        System.out.println("结束并行查询app首页信息,总耗时：" + (System.currentTimeMillis() - beginTime));        return buildResponse(userInfoDTO, bannerDTO, labelDTO);    }

基于以上代码，小伙伴们，是否还有其他方面的优化想法呢？ 比如这几个Callable查询任务，我们是不是也可以抽取一下？让代码更加简洁。

二话不说，现在我们直接建一个BaseTaskCommand类，实现Callable接口，把查询用户信息、查询banner信息、label标签信息的查询任务放进去。

代码如下：

public class BaseTaskCommand implements Callable<BaseRspDTO<Object>> {    private String key;    private AppInfoReq req;    private IUserService userService;    private IBannerService bannerService;    private ILabelService labelService;    public BaseTaskCommand(String key, AppInfoReq req, IUserService userService, IBannerService bannerService, ILabelService labelService) {        this.key = key;        this.req = req;        this.userService = userService;        this.bannerService = bannerService;        this.labelService = labelService;    }    @Override    public BaseRspDTO<Object> call() throws Exception {        if ("userInfoDTO".equals(key)) {            UserInfoParam userInfoParam = buildUserParam(req);            UserInfoDTO userInfoDTO = userService.queryUserInfo(userInfoParam);            BaseRspDTO<Object> userBaseRspDTO = new BaseRspDTO<Object>();            userBaseRspDTO.setKey("userInfoDTO");            userBaseRspDTO.setData(userInfoDTO);            return userBaseRspDTO;        } else if ("bannerDTO".equals(key)) {            BannerParam bannerParam = buildBannerParam(req);            BannerDTO bannerDTO = bannerService.queryBannerInfo(bannerParam);            BaseRspDTO<Object> bannerBaseRspDTO = new BaseRspDTO<Object>();            bannerBaseRspDTO.setKey("bannerDTO");            bannerBaseRspDTO.setData(bannerDTO);            return bannerBaseRspDTO;        } else if ("labelDTO".equals(key)) {            LabelParam labelParam = buildLabelParam(req);            LabelDTO labelDTO = labelService.queryLabelInfo(labelParam);            BaseRspDTO<Object> labelBaseRspDTO = new BaseRspDTO<Object>();            labelBaseRspDTO.setKey("labelDTO");            labelBaseRspDTO.setData(labelDTO);            return labelBaseRspDTO;        }        return null;    }    private UserInfoParam buildUserParam(AppInfoReq req) {        return new UserInfoParam();    }    private BannerParam buildBannerParam(AppInfoReq req) {        return new BannerParam();    }    private LabelParam buildLabelParam(AppInfoReq req) {        return new LabelParam();    }}

以上这块代码，构造函数还是有比较多的参数，并且call()方法中，有多个if...else...,如果新增一个分支（比如查询浮层信息），那又得在call方法里修改了，并且BaseTaskCommand的构造器也要修改了。

大家是否有印象，多程序中出现多个if...else...时，我们就可以考虑使用策略模式+工厂模式优化。

我们声明多个策略实现类，如下：

public interface IBaseTask {    //返回每个策略类的key，如    String getTaskType();    BaseRspDTO<Object> execute(AppInfoReq req);}//用户信息策略类@Servicepublic class UserInfoStrategyTask implements IBaseTask {    @Autowired    private IUserService userService;    @Override    public String getTaskType() {        return "userInfoDTO";    }    @Override    public BaseRspDTO<Object> execute(AppInfoReq req) {        UserInfoParam userInfoParam = userService.buildUserParam(req);        UserInfoDTO userInfoDTO = userService.queryUserInfo(userInfoParam);        BaseRspDTO<Object> userBaseRspDTO = new BaseRspDTO<Object>();        userBaseRspDTO.setKey(getTaskType());        userBaseRspDTO.setData(userBaseRspDTO);        return userBaseRspDTO;    }}/**  * banner信息策略实现类  **/@Servicepublic class BannerStrategyTask implements IBaseTask {    @Autowired    private IBannerService bannerService;    @Override    public String getTaskType() {        return "bannerDTO";    }    @Override    public BaseRspDTO<Object> execute(AppInfoReq req) {        BannerParam bannerParam = bannerService.buildBannerParam(req);        BannerDTO bannerDTO = bannerService.queryBannerInfo(bannerParam);        BaseRspDTO<Object> bannerBaseRspDTO = new BaseRspDTO<Object>();        bannerBaseRspDTO.setKey(getTaskType());        bannerBaseRspDTO.setData(bannerDTO);        return bannerBaseRspDTO;    }}...

然后这几个策略实现类，怎么交给spring管理呢？ 我们可以实现ApplicationContextAware接口，把策略的实现类注入到一个map，然后根据请求方不同的策略请求类型(即DTO的类型），去实现不同的策略类调用。其实这类似于工厂模式的思想。代码如下：

/**  * 策略工厂类  **/@Componentpublic class TaskStrategyFactory implements ApplicationContextAware {    private Map<String, IBaseTask> map = new ConcurrentHashMap<>();    @Override    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {        Map<String, IBaseTask> tempMap = applicationContext.getBeansOfType(IBaseTask.class);        tempMap.values().forEach(iBaseTask -> {            map.put(iBaseTask.getTaskType(), iBaseTask);        });    }    public BaseRspDTO<Object> executeTask(String key, AppInfoReq req) {        IBaseTask baseTask = map.get(key);        if (baseTask != null) {            System.out.println("工厂策略实现类执行");            return baseTask.execute(req);        }        return null;    }}

有了策略工厂类TaskStrategyFactory，我们再回来优化下BaseTaskCommand类的代码。它的构造器已经不需要多个IUserService userService, IBannerService bannerService, ILabelService labelService啦，只需要策略工厂类TaskStrategyFactory即可。同时策略也不需要多个if...else...判断了，用策略工厂类TaskStrategyFactory代替即可。优化后的代码如下：

ExecutorService executor = Executors.newFixedThreadPool(10);//查询用户信息CompletionService<UserInfoDTO> userDTOCompletionService = new ExecutorCompletionService<UserInfoDTO>(executor);Callable<UserInfoDTO> userInfoDTOCallableTask = () -> {   UserInfoParam userInfoParam = buildUserParam(req);   return userService.queryUserInfo(userInfoParam);};userDTOCompletionService.submit(userInfoDTOCallableTask);0

因此整个app首页信息并行查询，就可以优化成这样啦，如下：

ExecutorService executor = Executors.newFixedThreadPool(10);//查询用户信息CompletionService<UserInfoDTO> userDTOCompletionService = new ExecutorCompletionService<UserInfoDTO>(executor);Callable<UserInfoDTO> userInfoDTOCallableTask = () -> {   UserInfoParam userInfoParam = buildUserParam(req);   return userService.queryUserInfo(userInfoParam);};userDTOCompletionService.submit(userInfoDTOCallableTask);1

5. 思考总结

以上代码整体优化下来，已经很简洁啦。那还有没有别的优化思路呢。

其实还是有的，比如，把唯一标记的key定义为枚举，而不是写死的字符串"userInfoDTO"、"bannerDTO"，"labelDTO"。还有，除了CompletionService，有些小伙伴喜欢用CompletableFuture实行并行调用。

本文大家学到了哪些知识呢？

如何优化接口性能？某些场景下，可以使用并行调用代替串行。

如何实现并行调用呢？ 可以使用CompletionService。

学到的后端思维是？ 日常开发中，要学会抽取通用的方法、或者工具。

策略模式和工厂模式的应用

本文的话，设计模式这块还不是很详细，然后下一篇，给大家讲讲，我是如何在现有代码基础上，抽取设计模式的哈。然后，如果大家需要本文的完整代码的话，可以关注我的公众号：捡田螺的小男孩，里面有我的联系方式哈。

原文：https://juejin.cn/post/7098874204468019236