前言：
1、实现功能大致类似于“Callable和FutureTask的使用”
2、本文主要用于介绍实现CompletionService的利弊，请先了解FutureTask的相关知识
让我们比较一下FutureTask和CompletionService的区别。
代码如下所示。
publicclassmycompletionserviceimplementscallablestring {权限字符串用户名称； 私有长睡眠时间； publicmycompletionservice (string username，long sleepTime ) { this.userName=userName； this.sleepTime=sleepTime； } @ overridepublicstringcall (throws exception ) thread.sleep (sleep time )； 返回用户名称； } publicstaticvoidmain (string [ ] args ) throws InterruptedException，execution exception (mycompletionserviceservice1=new mycompletionserviceservice3=newmycompletionservice (' CCC '，3000 )； mycompletionserviceservice4=newmycompletionservice (' DDD '，2400 )； mycompletionserviceservice5=newmycompletionservice (' eee '，3600 )； listcallablelist=newarraylistcallable (； list.add(service1； list.add (服务2； list.add(service3； list.add(service4； list.add(service5； executorserviceexecutor=executors.newfixedthreadpool (10； //completionservicecompletionservicestringcompletionservice=newexecutorcompletionservice (执行程序)； system.out.println (start time : ) system.currenttimemillis ()； for(intI=0； i5； I )//提交对象为callable类型completion service.submit (list.get ) I )； }for(intI=0； i5； I ) ({ System.out.println ) (等待打印(() ) ) )第个返回值) )； system.out.println (' username : ' completion service.take ).get((-当前时间(System.currentTimeMillis ) ) ) ) )。 } system.out.println (end time : ) system.currenttimemillis ) )； //futuretaskfuturetaskstringtask1=newfuturetask (service1 )； futuretaskstringtask2=newfuturetask (service2 )； futuretaskstringtask3=newfuturetask (service 3； futuretaskstringtask4=newfuturetask (service 4； futuretaskstringtask5=newfuturetask (service 5； Lis
t<FutureTask> ftList = new ArrayList<>(); ftList.add(task1); ftList.add(task2); ftList.add(task3); ftList.add(task4); ftList.add(task5); System.out.println("startTime:"+System.currentTimeMillis()); for (int i = 0;i < 5;i++){ executor.execute(ftList.get(i)); } for (int i = 0;i < 5;i++) { System.out.println("等待打印第"+(i+1)+"个返回值"); System.out.println("userName:"+ftList.get(i).get()+"---"+System.currentTimeMillis()); } System.out.println("endTime:"+System.currentTimeMillis()); }}
上方的为CompletionService执行结果，下方为FutrueTask执行结果
其实从业务执行时间上看，消耗时间实质上差不多，但区别是：
1，FutureTask一定是按照调用get()方法的顺序执行的，但调用get()方法是会阻塞主线程，如果两个子线程第一个执行速度慢于第二个线程，那么输出的一定是同时输出。
2，CompletionService的则是按照线程先执行完的顺序输出的，改善了FutureTask的get()方法阻塞！。
下面执行一个业务系统的demo，其他类的代码在"Callable 和 FutureTask 的使用"一文中。
public Map<String,Object> sendMailAndMessage2(Map<String, Object> params){ MailCallable mailCallable = new MailCallable(params); MessageCallable messageCallable = new MessageCallable(params); List<Callable> list = new ArrayList<>(); list.add(mailCallable); list.add(messageCallable); // 创建线程池 ExecutorService executor = Executors.newFixedThreadPool(5); // 将线程池作为参数传入completionService中 CompletionService completionService = new ExecutorCompletionService(executor); Map<String,Object> map = new HashMap<String, Object>(); try { for (int i = 0; i < list.size(); i++) { completionService.submit(list.get(i)); } for (int i = 0; i < list.size(); i++) { map.putAll((Map<String, Object>)completionService.take().get()); } }catch (Exception e){ e.printStackTrace(); } return map;}
执行结果：
必要的五步：
//1、创建线程池（这里需要考虑线程池是单独给以下逻辑使用，还是全局使用的问题，单独使用放在方法中，全局使用放在外部）ThreadPoolExecutor executor = DefaultThreadPoolFactory.initThreadPool(5, 10);//2、创建异步执行器（VO只是一个实体类，没有做任何处理）CompletionService<VO> completionService = new ExecutorCompletionService(executor);//3、任务提交给异步执行器（这里submit的类对象必须实现Callable接口，此处用匿名内部类的方式new CallAble(){}}） completionService.submit(() -> { //业务代码 });//4、获取执行结果（submit了几个，获取几次）for (int i = 0; i < VOList.size(); i++) { VO vo = completionService.take().get(); if (null != vo){ dataList.add(vo); }}//5、关闭线程池executor.shutdown();
实际中应用例子(浙江省xxx项目中源代码）
// 创建线程池，有自己已经封装好的代码类ThreadPoolExecutor executor = DefaultThreadPoolFactory.initThreadPool(5, 10);// 创建异步执行器，泛型里写要操作的实体类CompletionService<VO> completionService = new ExecutorCompletionService(executor);// list中的vo都并发执行操作for (VO vo : baseVOList) { completionService.submit(() -> { //业务代码 });}// 获取每个线程执行的值for (int i = 0; i < baseVOList.size(); i++) { SentimentProcessBaseVO vo = completionService.take().get(); if (null != vo){ dataList.add(vo); }}// 关闭线程池executor.shutdown();