多任务线程池CompletionService使用详解

我会带着你远行 2022-02-24 02:16 318阅读 0赞

一、应用场景

当多个任务需要在线程池里执行时需要CompletionService

二、实战项目例子

比如有一个大bigList集合，需要根据大List集合内的每个元素去查询数据库，如果把这个大bigList一下放到线程池中执行，由于硬件设备cpu盒数、线程池配置的线程数等影响，可能执行效率并不是很高，但是如果把这个大bigList集合根据实际情况拆分成多个小littleList，把每个小littleList放到线程池去执行，这样会效率更高些。

三、代码示例

>     package com.lsl.threadpoolexample;
>     
>     import org.springframework.beans.factory.annotation.Autowired;
>     import org.springframework.stereotype.Service;
>     
>     import java.util.ArrayList;
>     import java.util.List;
>     import java.util.concurrent.*;
>     
>     /**
>      * @author: lsl
>      * @date: 2019/4/2
>      */
>     @Service("queryUserService")
>     public class QueryUserServiceImpl {
>     
>         @Autowired
>         QueryUserDao queryUserDao;
>     
>         public  List<UserPO> queryUser(List<String> bigIds) throws Exception{
>             //存储返回查询结果
>             List<UserPO> resultList = new ArrayList<>();
>             //创建线程池
>             ExecutorService executor = Executors.newFixedThreadPool(5);
>             CompletionService<List<UserPO>> completionService = new ExecutorCompletionService<>(executor);
>     
>             List<String> littleIds;
>             for (int i = 0; i<bigIds.size();i+=50){
>                 //拆分成多个任务，一个任务大小为50个元素
>                 if (i+50>bigIds.size()){
>                     littleIds = bigIds.subList(i,bigIds.size());
>                 }else {
>                     littleIds = bigIds.subList(i,i+50);
>                 }
>                 //每个拆分的任务放到线程池执行
>                 completionService.submit(new QueryUserTask(littleIds));
>             }
>     
>             for (int i = 0; i<bigIds.size();i+=50){
>                 //获取每个任务的执行结果
>                 List<UserPO> list = completionService.take().get();
>                 resultList.addAll(list);
>             }
>             return resultList;
>         }
>     
>         /**
>          * 查询任务
>          */
>         private class QueryUserTask implements Callable<List<UserPO>> {
>             private List<String> ids;
>     
>             public QueryUserTask(List<String> ids){
>                 this.ids = ids;
>             }
>             @Override
>             public List<UserPO> call() throws Exception {
>                 List<UserPO> list = queryUserDao.queryUserById(ids);
>                 return list;
>             }
>     
>         }
>     }

四、注意事项

CompletionService.take().get()是获取线程执行的结果，这个结果和每个任务实际执行顺序无关，是先获取到执行完的线程结果，如果没有一个线程执行完成，则是阻塞状态。