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

缺乏、安全感 2022-05-18 09:16 226阅读 0赞

《Java并发编程实践》一书6.3.5节CompletionService：Executor和BlockingQueue，有这样一段话：

"如果向Executor提交了一组计算任务，并且希望在计算完成后获得结果，那么可以保留与每个任务关联的Future，然后反复使用get方法，同时将参数timeout指定为0，从而通过轮询来判断任务是否完成。这种方法虽然可行，但却有些繁琐。幸运的是，还有一种更好的方法：完成服务CompletionService。"

这是什么意思呢？我们通过一个例子，分别使用繁琐的做法和CompletionService来完成，清晰的对比能让我们更好的理解上面的一段话和CompletionService这个API提供的初衷。考虑这样的场景，有5个Callable任务分别返回5个整数，然后我们在main方法中按照各个任务完成的先后顺序，在控制台打印返回结果。

package net.aty.completeservice;
    
    import java.util.concurrent.Callable;
    import java.util.concurrent.TimeUnit;
    
    public class ReturnAfterSleepCallable implements Callable<Integer>
    {
    	private int sleepSeconds;
    
    	private int returnValue;
    
    	public ReturnAfterSleepCallable(int sleepSeconds, int returnValue)
    	{
    		this.sleepSeconds = sleepSeconds;
    		this.returnValue = returnValue;
    	}
    
    	@Override
    	public Integer call() throws Exception
    	{
    		System.out.println("begin to execute.");
    
    		TimeUnit.SECONDS.sleep(sleepSeconds);
    
    		System.out.println("end to execute.");
    
    		return returnValue;
    	}
    }

这个任务会接受2个参数，睡眠指定的时间后，返回指定的结果。睡眠时间越短，意味着任务越先执行完成。

# 1.繁琐的做法 #

通过一个List来保存每个任务返回的Future，然后轮询这些Future，直到每个Future都已完成。我们不希望出现因为排在前面的任务阻塞导致后面先完成的任务的结果没有及时获取的情况，所以在调用get方式时，需要将超时时间设置为0。

package net.aty.completeservice;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.Future;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.TimeoutException;
    
    public class TraditionalTest
    {
    	public static void main(String[] args)
    	{
    		int taskSize = 5;
    
    		ExecutorService executor = Executors.newFixedThreadPool(taskSize);
    
    		List<Future<Integer>> futureList = new ArrayList<Future<Integer>>();
    
    		for (int i = 1; i <= taskSize; i++)
    		{
    			int sleep = taskSize - i; // 睡眠时间
    
    			int value = i; // 返回结果
    
    			// 向线程池提交任务
    			Future<Integer> future = executor.submit(new ReturnAfterSleepCallable(sleep, value));
    			
    			// 保留每个任务的Future
    			futureList.add(future);
    		}
    		
    		// 轮询,获取完成任务的返回结果
    		while(taskSize > 0)
    		{
    			for (Future<Integer> future : futureList)
    			{
    				Integer result = null;
    				
    				try
    				{
    					result = future.get(0, TimeUnit.SECONDS);
    				} catch (InterruptedException e)
    				{
    					e.printStackTrace();
    				} catch (ExecutionException e)
    				{
    					e.printStackTrace();
    				} catch (TimeoutException e)
    				{
    					// 超时异常需要忽略,因为我们设置了等待时间为0,只要任务没有完成,就会报该异常
    				}
    				
    				// 任务已经完成
    				if(result != null)
    				{
    					System.out.println("result=" + result);
    					
    					// 从future列表中删除已经完成的任务
    					futureList.remove(future);  
    					taskSize--;
    					//此处必须break，否则会抛出并发修改异常。（也可以通过将futureList声明为CopyOnWriteArrayList类型解决） 
    					break; // 进行下一次while循环
    				}
    			}
    		}
    		
    		// 所有任务已经完成,关闭线程池
    		System.out.println("all over.");
    		executor.shutdown();
    	}
    	
    	
    	
    }

可见轮询future列表非常的复杂，而且还有很多异常需要处理，TimeOutException异常需要忽略；还要通过双重循环和break，防止遍历集合的过程中，出现并发修改异常。这么多需要考虑的细节，程序员很容易犯错。

# 2.使用CompletionService #

package net.aty.completeservice;
    
    import java.util.concurrent.CompletionService;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ExecutorCompletionService;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    public class CompletionServiceTest
    {
    	public static void main(String[] args)
    	{
    		int taskSize = 5;
    
    		ExecutorService executor = Executors.newFixedThreadPool(taskSize);
    
    		// 构建完成服务
    		CompletionService<Integer> completionService = new ExecutorCompletionService<Integer>(
    				executor);
    
    		for (int i = 1; i <= taskSize; i++)
    		{
    			int sleep = taskSize - i; // 睡眠时间
    
    			int value = i; // 返回结果
    
    			// 向线程池提交任务
    			completionService
    					.submit(new ReturnAfterSleepCallable(sleep, value));
    		}
    
    		// 按照完成顺序,打印结果
    		for (int i = 0; i < taskSize; i++)
    		{
    			try
    			{
    				System.out.println(completionService.take().get());
    			} catch (InterruptedException e)
    			{
    				e.printStackTrace();
    			} catch (ExecutionException e)
    			{
    				e.printStackTrace();
    			}
    		}
    
    		// 所有任务已经完成,关闭线程池
    		System.out.println("all over.");
    		executor.shutdown();
    	}
    }

可见使用CompletionService不会有TimeOutExeception的问题，不用遍历future列表，不用担心并发修改异常。

# 3.CompletionService和ExecutorCompletionService的实现 #

JDK源码中CompletionService的javadoc说明如下：

/**
     * A service that decouples the production of new asynchronous tasks
     * from the consumption of the results of completed tasks.  Producers
     * <tt>submit</tt> tasks for execution. Consumers <tt>take</tt>
     * completed tasks and process their results in the order they
     * complete. 
     */

也就是说，CompletionService实现了生产者提交任务和消费者获取结果的解耦，生产者和消费者都不用关心任务的完成顺序，由CompletionService来保证，消费者一定是按照任务完成的先后顺序来获取执行结果。

ExecutorCompletionService是CompletionService的实现，融合了线程池Executor和阻塞队列BlockingQueue的功能。

public ExecutorCompletionService(Executor executor) {
            if (executor == null)
                throw new NullPointerException();
            this.executor = executor;
            this.aes = (executor instanceof AbstractExecutorService) ?
                (AbstractExecutorService) executor : null;
            this.completionQueue = new LinkedBlockingQueue<Future<V>>();
        }

到这里可以推测，按照任务的完成顺序获取结果，就是通过阻塞队列实现的，阻塞队列刚好具有这样的性质：阻塞和有序。

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

public Future<V> submit(Callable<V> task) {
            if (task == null) throw new NullPointerException();
            RunnableFuture<V> f = newTaskFor(task);
            executor.execute(new QueueingFuture(f));
            return f;
    }

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

/**
      * FutureTask extension to enqueue upon completion
      */
    private class QueueingFuture extends FutureTask<Void> {
            QueueingFuture(RunnableFuture<V> task) {
                super(task, null);
                this.task = task;
            }
            protected void done() { completionQueue.add(task); }
            private final Future<V> task;
    }

这里简单说明下：FutureTask.done()，这个方法默认什么都不做，就是一个回调，当提交的线程池中的任务完成时，会被自动调用。这也就说时候，当任务完成的时候，会自动执行QueueingFuture.done()方法，将返回结果加入到阻塞队列中，加入的顺序就是任务完成的先后顺序。

至此CompletionService和ExecutorCompletionService的讲解结束。特别感谢http://xw-z1985.iteye.com/blog/1997077这篇博客的作者**海浪儿**，解决了我在看《java并发编程实践》中的困惑，本文也基本上是照抄那篇博文的。