ExecutorCompletionService分析及使用

短命女 2022-02-24 04:09 184阅读 0赞

当我们通过Executor提交一组并发执行的任务，并且希望在每一个任务完成后能立即得到结果，有两种方式可以采取：

方式一：

通过一个list来保存一组future，然后在循环中轮训这组future,直到每个future都已完成。如果我们不希望出现因为排在前面的任务阻塞导致后面先完成的任务的结果没有及时获取的情况，那么在调用get方式时，需要将超时时间设置为0

**Java代码  ![收藏代码][icon_star.png]**

1.  **public** **class** CompletionServiceTest \{  
2.    
3.      **static** **class** Task **implements** Callable<String>\{  
4.          **private** **int** i;  
5.            
6.          **public** Task(**int** i)\{  
7.              **this**.i = i;  
8.          \}  
9.    
10.         @Override  
11.         **public** String call() **throws** Exception \{  
12.             Thread.sleep(10000);  
13.             **return** Thread.currentThread().getName() + "执行完任务：" + i;  
14.         \}     
15.     \}  
16.       
17.     **public** **static** **void** main(String\[\] args)\{  
18.         testUseFuture();  
19.     \}  
20.       
21.     **private** **static** **void** testUseFuture()\{  
22.         **int** numThread = 5;  
23.         ExecutorService executor = Executors.newFixedThreadPool(numThread);  
24.         List<Future<String>> futureList = **new** ArrayList<Future<String>>();  
25.         **for**(**int** i = 0;i<numThread;i++ )\{  
26.             Future<String> future = executor.submit(**new** CompletionServiceTest.Task(i));  
27.             futureList.add(future);  
28.         \}  
29.                   
30.         **while**(numThread > 0)\{  
31.             **for**(Future<String> future : futureList)\{  
32.                 String result = **null**;  
33.                 **try** \{  
34.                     result = future.get(0, TimeUnit.SECONDS);  
35.                 \} **catch** (InterruptedException e) \{  
36.                     e.printStackTrace();  
37.                 \} **catch** (ExecutionException e) \{  
38.                     e.printStackTrace();  
39.                 \} **catch** (TimeoutException e) \{  
40.                     //超时异常直接忽略  
41.                 \}  
42.                 **if**(**null** != result)\{  
43.                     futureList.remove(future);  
44.                     numThread--;  
45.                     System.out.println(result);  
46.                     //此处必须break，否则会抛出并发修改异常。（也可以通过将futureList声明为CopyOnWriteArrayList类型解决）  
47.                     **break**;  
48.                 \}  
49.             \}  
50.         \}  
51.     \}  
52. \}

方式二：

第一种方式显得比较繁琐，通过使用ExecutorCompletionService，则可以达到代码最简化的效果。

**Java代码  ![收藏代码][icon_star.png]**

1.  **public** **class** CompletionServiceTest \{  
2.    
3.      **static** **class** Task **implements** Callable<String>\{  
4.          **private** **int** i;  
5.            
6.          **public** Task(**int** i)\{  
7.              **this**.i = i;  
8.          \}  
9.    
10.         @Override  
11.         **public** String call() **throws** Exception \{  
12.             Thread.sleep(10000);  
13.             **return** Thread.currentThread().getName() + "执行完任务：" + i;  
14.         \}     
15.     \}  
16.       
17.     **public** **static** **void** main(String\[\] args) **throws** InterruptedException, ExecutionException\{  
18.         testExecutorCompletionService();  
19.     \}  
20.       
21.     **private** **static** **void** testExecutorCompletionService() **throws** InterruptedException, ExecutionException\{  
22.         **int** numThread = 5;  
23.         ExecutorService executor = Executors.newFixedThreadPool(numThread);  
24.         CompletionService<String> completionService = **new** ExecutorCompletionService<String>(executor);  
25.         **for**(**int** i = 0;i<numThread;i++ )\{  
26.             completionService.submit(**new** CompletionServiceTest.Task(i));  
27.         \}  
28. \}  
29.           
30.         **for**(**int** i = 0;i<numThread;i++ )\{       
31.             System.out.println(completionService.take().get());  
32.         \}  
33.           
34.     \}

ExecutorCompletionService分析：

CompletionService是Executor和BlockingQueue的结合体。

**Java代码  ![收藏代码][icon_star.png]**

1.  **public** ExecutorCompletionService(Executor executor) \{  
2.          **if** (executor == **null**)  
3.              **throw** **new** NullPointerException();  
4.          **this**.executor = executor;  
5.          **this**.aes = (executor **instanceof** AbstractExecutorService) ?  
6.              (AbstractExecutorService) executor : **null**;  
7.          **this**.completionQueue = **new** LinkedBlockingQueue<Future<V>>();  
8.      \}

任务的提交和执行都是委托给Executor来完成。当提交某个任务时，该任务首先将被包装为一个QueueingFuture，

**Java代码  ![收藏代码][icon_star.png]**

1.  **public** Future<V> submit(Callable<V> task) \{  
2.          **if** (task == **null**) **throw** **new** NullPointerException();  
3.          RunnableFuture<V> f = newTaskFor(task);  
4.          executor.execute(**new** QueueingFuture(f));  
5.          **return** f;  
6.      \}

QueueingFuture是FutureTask的一个子类，通过改写该子类的done方法，可以实现当任务完成时，将结果放入到BlockingQueue中。

**Java代码  ![收藏代码][icon_star.png]**

1.  **private** **class** QueueingFuture **extends** FutureTask<Void> \{  
2.          QueueingFuture(RunnableFuture<V> task) \{  
3.              **super**(task, **null**);  
4.              **this**.task = task;  
5.          \}  
6.          **protected** **void** done() \{ completionQueue.add(task); \}  
7.          **private** **final** Future<V> task;  
8.      \}

而通过使用BlockingQueue的take或poll方法，则可以得到结果。在BlockingQueue不存在元素时，这两个操作会阻塞，一旦有结果加入，则立即返回。

**Java代码  ![收藏代码][icon_star.png]**

1.  **public** Future<V> take() **throws** InterruptedException \{  
2.      **return** completionQueue.take();  
3.  \}  
4.    
5.  **public** Future<V> poll() \{  
6.      **return** completionQueue.poll();  
7.  \}

[icon_star.png]: /images/20220224/25e5a57aec3145f68b348fafc11d9066.png

发表评论取消回复

表情：

评论列表（有 0 条评论，184人围观）

还没有评论，来说两句吧...

相关阅读

相关 CompletionService和ExecutorCompletionService详解

、CompletionService接口定义了一组任务管理接口: * submit() - 提交任务 * take() - 获取任务结果 * poll() ...

旧城等待，/ 2024年04月17日 06:04/ 0 赞/ 50 阅读

相关超实用高并发编程ExecutorCompletionService案例分析与源码解读

要点解说假设现在有一大批需要进行计算的任务，为了提高整批任务的执行效率，你可能会使用线程池，向线程池中不断submit异步计算任务，同时你需要保留与每个任务关联的Fut

￡神魔★判官ぃ/ 2024年03月23日 14:49/ 0 赞/ 50 阅读

相关 ExecutorCompletionService

`ExecutorCompletionService介绍` `ExecutorCompletionService` 是 Java 提供的一个用于并发处理任务的类。它的作用

青旅半醒/ 2023年10月16日 09:54/ 0 赞/ 30 阅读

相关 Java ExecutorCompletionService类

`一、为什么使用ExecutorCompletionService` > 假设现在有一大批需要进行计算的任务，为了提高整批任务的执行效率，你可能会使用线程池，向线程池中不

「爱情、让人受尽委屈。」/ 2023年10月08日 12:00/ 0 赞/ 7 阅读

相关 @EnableTransactionManagement的使用及分析

使用spring事务的时候有一个总开关@EnableTransactionManagement 1. 事务管理器 Spring Boot 使用事务非常简单 1. 使

梦里梦外;/ 2022年11月17日 14:45/ 0 赞/ 191 阅读

相关 ExecutorService 和 ExecutorCompletionService 使用区别及ExecutorCompletionService的应用场景

文章目录案例使用 ExecutorService + Future 使用CompletionService实现

男娘i/ 2022年11月11日 14:56/ 0 赞/ 180 阅读

相关 JUC ExecutorCompletionService 分析

基本介绍 `ExecutorCompletionService`，JUC中提供的`CompletionService`接口实现，用于实现有限任务的执行以及任务结果的有序返

你的名字/ 2022年05月19日 04:44/ 0 赞/ 192 阅读

相关 JDK并发工具包CompletionService和ExecutorCompletionService的好处和使用场景

《Java并发编程实践》一书6.3.5节CompletionService：Executor和BlockingQueue，有这样一段话： "如果向Executor提交了一

缺乏、安全感/ 2022年05月18日 09:16/ 0 赞/ 180 阅读

相关 ExecutorCompletionService

java.util.concurrent 类 ExecutorCompletionService<V> java.lang.Object

快来打我*/ 2022年04月23日 09:54/ 0 赞/ 151 阅读

相关 ExecutorCompletionService分析及使用

当我们通过Executor提交一组并发执行的任务，并且希望在每一个任务完成后能立即得到结果，有两种方式可以采取：方式一：通过一个list来保存一组future，然

短命女/ 2022年02月24日 04:09/ 0 赞/ 185 阅读