亲宝软件园·资讯

展开

Java并行调用模板

​ 捡田螺的小男孩   ​ 人气:0

前言:

本文主要介绍内容有:

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);
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);
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;
    }

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

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

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);
}
//用户信息策略类
@Service
public 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信息策略实现类
  **/
@Service
public 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的类型),去实现不同的策略类调用。其实这类似于工厂模式的思想。

代码如下:

/**
  * 策略工厂类
  **/
@Component
public 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代替即可。

优化后的代码如下:

public class BaseTaskCommand implements Callable<BaseRspDTO<Object>> {
    private String key;
    private AppInfoReq req;
    private TaskStrategyFactory taskStrategyFactory;
    public BaseTaskCommand(String key, AppInfoReq req, TaskStrategyFactory taskStrategyFactory) {
        this.key = key;
        this.req = req;
        this.taskStrategyFactory = taskStrategyFactory;
    }
    @Override
    public BaseRspDTO<Object> call() throws Exception {
        return taskStrategyFactory.executeTask(key, req);
    }
}  

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

public AppHeadInfoResponse parallelQueryAppHeadPageInfo2(AppInfoReq req) {
    long beginTime = System.currentTimeMillis();
    System.out.println("开始并行查询app首页信息(最终版本),开始时间:" + beginTime);
    List<Callable<BaseRspDTO<Object>>> taskList = new ArrayList<>();
    //用户信息查询任务
    taskList.add(new BaseTaskCommand("userInfoDTO", req, taskStrategyFactory));
    //banner查询任务
    taskList.add(new BaseTaskCommand("bannerDTO", req, taskStrategyFactory));
    //标签查询任务
    taskList.add(new BaseTaskCommand("labelDTO", req, taskStrategyFactory));

    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 (BaseRspDTO<Object> baseRspDTO : resultList) {
        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);
  }

5. 思考总结

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

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

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

加载全部内容

相关教程
猜你喜欢
用户评论