如何写一个并行调用模版
后端思维篇:手把手教你写一个并行调用模板
- 一个串行调用的例子(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`实现并行调用的例子已经实现啦。是不是很开心,哈哈。
到这里为止,一个基于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,如是usetInfoDTO还是bannerDTO,还是labelDTO
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
,然后根据请求方不同的策略请求类型(即userInfoDTO还是bannerDTO等),去实现不同的策略类调用。其实这类似于工厂模式的思想。代码如下:
/**
* 策略工厂类
**/
@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
实行并行调用,大家可以自己动手操戈写一写。
本文大家学到了哪些知识呢?
- 如何优化接口性能?某些场景下,可以使用并行调用代替串行。
- 如何实现并行调用呢?可以使用
CompletionService
。 - 学到的后端思维是?日常开发中,要学会抽取通用的方法、或者工具。
- 策略模式和工厂模式的应用
Enjoy Reading This Article?
Here are some more articles you might like to read next: