CompletionService: 擷取異步任務執行結果

**1、ExecutorService **

使用Future和Callable可以擷取線程池中任務執行的結果，但其擷取方式是阻塞的，根據添加到線程池中的線程順序依次擷取，擷取不到就阻塞住主線程。

public class AsyncTask {
	public static void main(String[] args) {
		ExecutorService executor = Executors.newFixedThreadPool(3);
		List<Future<Integer>> futures = new ArrayList<>();
		for (int i = 5; i > 2; i--) {
			final int time = i;
			futures.add(executor.submit(new Callable<Integer>() {

				@Override
				public Integer call() throws Exception {
					TimeUnit.SECONDS.sleep(time);
					return time;
				}
			}));
		}
		
		for (Future<Integer> future : futures) {
			try {
				System.out.println(future.get());
			} catch (InterruptedException | ExecutionException e) {
				e.printStackTrace();
			}
		}
	}
}
           

執行結果：

5
4
3
           

這種方式有一個值得注意的地方是當首先擷取的任務結果耗時很長時，那麼即使耗時很短的任務的結果也無法先獲得，因為主線程阻塞在耗時長任務的get()方法處。

2、CompletionService

先看下CompletionService的實作方式：

public class AsyncTaskByCompletionService {
	public static void main(String[] args) {
		ExecutorService executor = Executors.newFixedThreadPool(3);
		CompletionService<Integer> completionService = new ExecutorCompletionService<>(executor);
		for (int i = 5; i > 2; i--) {
			final int time = i;
			completionService.submit(new Callable<Integer>() {
				@Override
				public Integer call() throws Exception {
					TimeUnit.SECONDS.sleep(time);
					return time;
				}
			});
		}
		
		for (int i = 0; i < 3; i++) {
			try {
				System.out.println(completionService.take().get());
			} catch (InterruptedException | ExecutionException e) {
				e.printStackTrace();
			}
		}
	}
}
           

執行結果：

3
4
5
           

CompletionService 可以先擷取耗時短任務的執行結果

3、CompletionService實作分析：

CompletionService将線程池Executor和阻塞隊列的功能融合在一起了，其内部維護了一個阻塞隊列，當任務執行完後就會把任務執行的結果Future加入到阻塞隊列中，任務還是由線程池中的線程處理。

CompletionService 接口的實作類是 ExecutorCompletionService，這個實作類的構造方法有兩個，分别是：

• 1.ExecutorCompletionService(Executor executor)
• 2.ExecutorCompletionService(Executor executor, BlockingQueue<Future> completionQueue)

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>>();
    }
           

構造法方法主要初始化了一個阻塞隊列，用來存儲已完成的task任務。

接下來看submit()方法：

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;
    }
           

submit()方法将任務包裝為QueueingFuture，其為FutureTask的子類，是以最終執行的是FutureTask的run()方法。

public void run() {
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return;
        try {
            Callable<V> c = callable;
            if (c != null && state == NEW) {
                V result;
                boolean ran;
                try {
                    result = c.call();
                    ran = true;
                } catch (Throwable ex) {
                    result = null;
                    ran = false;
                    setException(ex);
                }
                if (ran)
                    set(result);
            }
        } finally {
            // runner must be non-null until state is settled to
            // prevent concurrent calls to run()
            runner = null;
            // state must be re-read after nulling runner to prevent
            // leaked interrupts
            int s = state;
            if (s >= INTERRUPTING)
                handlePossibleCancellationInterrupt(s);
        }
    }
           

可以看到在該 FutureTask 中執行run方法，最終回調自定義的callable中的call方法，執行結束之後，通過 set(result) 處理執行結果：

protected void set(V v) {
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = v;
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
            finishCompletion();
        }
    }
           

繼續跟進finishCompletion()方法，在該方法中找到 done()方法，将會調用子類QueueingFuture的done()方法，将Future task添加到阻塞隊列中：

protected void done() { completionQueue.add(task); }
        private final Future<V> task;
           

隻要隊列中有元素，就可以調用阻塞隊列取元素API(poll取不到時傳回null，take取不到時阻塞，poll(time,unit)逾時取不到傳回null)得到任務執行結果。

但是，歸根結底CompletionService擷取異步任務執行結果當遇到耗時較長的任務時還是會阻塞。

netty提供了一種将Future添加listener，通過回調的方式實作非阻塞機制：

ChannelFuture future = channel.connect(new InetSocketAddress("127.0.0.1", 8080));
        //注冊到ChannelFutureListener，以便在操作完成時擷取通知
        future.addListener(new ChannelFutureListener()
        {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception
            {
                //檢查操作狀态
                if(future.isSuccess()){
                    //如果操作成功則建立一個ByteBuf以持有資料
                    ByteBuf buffer = Unpooled.copiedBuffer("Hello", Charset.defaultCharset());
                    //将資料異步的發送給用戶端節點
                    future.channel().writeAndFlush(buffer);
                }else{
                    System.out.println("connect failed.");
                }
            }
        });
           

總結：

• 1、ExecutorService送出異步任務在擷取傳回值時，前面耗時較長的任務會阻塞後面耗時較短任務結果的擷取
• 2、CompletionService送出的異步任務将Future結果存放在阻塞隊列中，耗時較短的任務能夠将結果首先加入阻塞隊列，這樣就可以及時擷取已執行完任務的結果做後續的處理
• 3、ExecutorService和CompletionService的方式對于耗時較長的任務結果的擷取都會阻塞，netty提供了一種回調通知的機制，實作了結果的非阻塞擷取