Java多執行緒6 CompletionService
CompletionService
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(); } } } }
CompletionService總結
相比ExecutorService,CompletionService可以更精確和簡便地完成非同步任務的執行
CompletionService的一個實現是ExecutorCompletionService,它是Executor和BlockingQueue功能的融合體,Executor完成計算任務,BlockingQueue負責儲存非同步任務的執行結果
在執行大量相互獨立和同構的任務時,可以使用CompletionService
CompletionService可以為任務的執行設定時限,主要是通過BlockingQueue的poll(long time,TimeUnit unit)為任務執行結果的取得限制時間,如果沒有完成就取消任務