Java多線程 CompletionService

Posted on by

1 CompletionService介紹

CompletionService用於提交一組Callable任務，其take方法返回已完成的一個Callable任務對應的Future對象。
如果你向Executor提交瞭一個批處理任務，並且希望在它們完成後獲得結果。為此你可以將每個任務的Future保存進一個集合，然後循環這個集合調用Future的get()取出數據。幸運的是CompletionService幫你做瞭這件事情。
CompletionService整合瞭Executor和BlockingQueue的功能。你可以將Callable任務提交給它去執行，然後使用類似於隊列中的take和poll方法，在結果完整可用時獲得這個結果，像一個打包的Future。
CompletionService的take返回的future是哪個先完成就先返回哪一個，而不是根據提交順序。

2 CompletionService源碼分析

首先看一下 構造方法：

   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任務。

然後看一下 completionService.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;
    }

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

可以看到，callable任務被包裝成QueueingFuture，而 QueueingFuture是 FutureTask的子類，所以最終執行瞭FutureTask中的run()方法。

來看一下該方法：

 public void run() {
 //判斷執行狀態，保證callable任務隻被運行一次
    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 {
            //這裡回調我們創建的callable對象中的call方法
                result = c.call();
                ran = true;
            } catch (Throwable ex) {
                result = null;
                ran = false;
                setException(ex);
            }
            if (ran)
            //處理執行結果
                set(result);
        }
    } finally {
        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) 處理執行結果：

    /**
     * Sets the result of this future to the given value unless
     * this future has already been set or has been cancelled.
     *
     * <p>This method is invoked internally by the {@link #run} method
     * upon successful completion of the computation.
     *
     * @param v the value
     */
    protected void set(V v) {
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = v;
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
            finishCompletion();
        }
    }

繼續跟進finishCompletion()方法，在該方法中找到 done()方法：

protected void done() { completionQueue.add(task); }

可以看到該方法隻做瞭一件事情，就是將執行結束的task添加到瞭隊列中，隻要隊列中有元素，我們調用take()方法時就可以獲得執行的結果。
到這裡就已經清晰瞭，異步非阻塞獲取執行結果的實現原理其實就是通過隊列來實現的，FutureTask將執行結果放到隊列中，先進先出，線程執行結束的順序就是獲取結果的順序。

CompletionService實際上可以看做是Executor和BlockingQueue的結合體。CompletionService在接收到要執行的任務時，通過類似BlockingQueue的put和take獲得任務執行的結果。CompletionService的一個實現是ExecutorCompletionService，ExecutorCompletionService把具體的計算任務交給Executor完成。

在實現上，ExecutorCompletionService在構造函數中會創建一個BlockingQueue（使用的基於鏈表的無界隊列LinkedBlockingQueue），該BlockingQueue的作用是保存Executor執行的結果。當計算完成時，調用FutureTask的done方法。當提交一個任務到ExecutorCompletionService時，首先將任務包裝成QueueingFuture，它是FutureTask的一個子類，然後改寫FutureTask的done方法，之後把Executor執行的計算結果放入BlockingQueue中。

QueueingFuture的源碼如下：

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

3 CompletionService實現任務

public class CompletionServiceTest {
    public static void main(String[] args) {

        ExecutorService threadPool = Executors.newFixedThreadPool(10);
        CompletionService<Integer> completionService = new ExecutorCompletionService<Integer>(threadPool);
        for (int i = 1; i <=10; i++) {
            final int seq = i;
            completionService.submit(new Callable<Integer>() {
                @Override
                public Integer call() throws Exception {

                    Thread.sleep(new Random().nextInt(5000));

                    return seq;
                }
            });
        }
        threadPool.shutdown();
        for (int i = 0; i < 10; i++) {
            try {
                System.out.println(
                        completionService.take().get());
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }

    }
}

7
3
9
8
1
2
4
6
5
10

4 CompletionService總結

相比ExecutorService，CompletionService可以更精確和簡便地完成異步任務的執行
CompletionService的一個實現是ExecutorCompletionService，它是Executor和BlockingQueue功能的融合體，Executor完成計算任務，BlockingQueue負責保存異步任務的執行結果
在執行大量相互獨立和同構的任務時，可以使用CompletionService
CompletionService可以為任務的執行設置時限，主要是通過BlockingQueue的poll(long time,TimeUnit unit)為任務執行結果的取得限制時間，如果沒有完成就取消任務

到此這篇關於Java多線程 CompletionService的文章就介紹到這瞭,更多相關Java多線程CompletionService內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

推薦閱讀：