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

男娘i 2022-11-11 14:56 208阅读 0赞

### 文章目录 ###

*   *  案例
     *   *  使用 ExecutorService + Future
         *  使用CompletionService实现
         *   *  CompletionService 类结构
             *  CompletionService构造方法
             *  实现和CompletionService一样的案例
     *  CompletionService的应用场景

## 案例 ##

假设我们用实现这样一种需求，并行去执行5个任务，然后获取任务结果去执行其他操作，你要如何编码实现？

### 使用 ExecutorService + Future ###

AtomicInteger atomicInteger = new AtomicInteger(1);
            ExecutorService executor = Executors.newFixedThreadPool(5);
            List<Future<Integer>> futures = new ArrayList<>();
            for (int i = 0; i < 10; i++) { 
            	// 这里使用了lambda 如果看不懂可以使用以前的实现方式
                Future<Integer> future = executor.submit(atomicInteger::getAndIncrement);
                futures.add(future);
            }
            for (Future<Integer> future : futures) { 
                // 获取会阻塞
                System.out.println(future.get(5, TimeUnit.SECONDS));
            }

上面的`executor.submit(atomicInteger::getAndIncrement);`  
使用了lambda 如果看不懂可以使用以前的实现方式 创建一个类实现Callable接口

static class Task implements Callable<Integer>{ 
            AtomicInteger atomicInteger = new AtomicInteger(1);
            @Override
            public Integer call() { 
                return atomicInteger.getAndIncrement();
            }
        }
        Future<Integer> future = executor.submit(new Task());

上面的实现方式有什么缺点呢？

1.  编码麻烦，需要用一个List去存储所有的Future， 如果前面的任务耗时很长，就会阻塞后面Future的执行，因为前面的Future.get() 会阻塞住

### 使用CompletionService实现 ###

#### CompletionService 类结构 ####

![在这里插入图片描述][20210321124454820.png]  
可以看到CompletionService 接口只有唯一一个实现类就是 ExecutorCompletionService

#### CompletionService构造方法 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70]  
可以看到ExecutorCompletionService的构造方法只有两个，一个是传入一个线程池，还要一个就是传入一个阻塞队列。

#### 实现和CompletionService一样的案例 ####

CompletionService<Integer> completionService = new ExecutorCompletionService<>(executor);
            for (int i = 0; i < 10; i++) { 
                completionService.submit(atomicInteger::getAndIncrement);
            }
            for (int i = 0; i < 10; i++) { 
                // 如果队列为空，那么调用 take() 方法的线程会被阻塞
                System.out.println(completionService.take().get(5, TimeUnit.SECONDS));
            }

可以看到使用CompletionService 编码相对来说简单一点。其次使用take()方法不会因为某个任务执行时间长而阻塞，所有任务的结果是谁先执行完先放入队列中，先获取。

缺点也是很明显的，每个任务结果都需要新开一个FutureTask去保存，非常消耗资源  
![在这里插入图片描述][20210321125714753.png]

## CompletionService的应用场景 ##

假设我们有这样一个需求，现在有A系统，部署了三个节点。我们需要获取A系统的一个接口数据，需要最快返回返回结果，而不是至少单纯调用某个节点，在某个节点调用不通再去掉用另外一个节点。这种需求如何实现呢？这就是类似Dubbo 中的 Forking Cluster(集群模式,并行地调用多个查询服务，只要有一个成功返回结果，整个服务就可以返回)。

其实实现起来也比较简单，我们只需要使用一个线程池同时去调用三个节点，哪个节点获取到了任务就直接返回，并停止其他任务即可。代码如下

ExecutorService threadPool = Executors.newFixedThreadPool(3);
            CompletionService<Integer> cs = new ExecutorCompletionService<>(threadPool);
            List<Future<Integer>> futures = new ArrayList<>(3);
            // 提交异步任务，保存 future到 futures
            futures.add(cs.submit(() -> { 
                TimeUnit.SECONDS.sleep(2);
                return 2;
            }));
    
            futures.add(cs.submit(() -> { 
                TimeUnit.SECONDS.sleep(3);
                return 3;
            }));
            futures.add(cs.submit(() -> 1));
            Integer result = 0;
            try { 
                for (int i = 0; i < 3; i++) { 
                    result = cs.take().get();
                    if (result != null) { 
                        break;
                    }
                }
            } catch (Exception e) { 
                e.printStackTrace();
            } finally { 
                //取消所有任务
                futures.forEach(f -> f.cancel(true));
            }
            // 输出结果
            System.out.println(result);
            threadPool.shutdown();

上面代码我们故意让前两个任务 休眠2-3秒，模拟网络延迟，而执行结果很明显也是0延时返回的，因为任务三线程没有休眠。

[20210321124454820.png]: /images/20221022/5bfbcf7803f14869b657b3c7ec01c816.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70]: /images/20221022/b93d1c5482044f588cd7fe2480b3e750.png
[20210321125714753.png]: /images/20221022/4d78ff673a7b4d52b3ccec806d3b8465.png