java.util.concurrent.ExecutorCompletionService 原始碼
執行緒池相關
原始碼:
package java.util.concurrent; public class ExecutorCompletionService<V> implements CompletionService<V> { private final Executor executor; private final AbstractExecutorService aes; private final BlockingQueue<Future<V>> completionQueue; 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; } private RunnableFuture<V> newTaskFor(Callable<V> task) { if (aes == null) return new FutureTask<V>(task); else return aes.newTaskFor(task); } private RunnableFuture<V> newTaskFor(Runnable task, V result) { if (aes == null) return new FutureTask<V>(task, result); else return aes.newTaskFor(task, result); } 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>>(); } public ExecutorCompletionService(Executor executor, BlockingQueue<Future<V>> completionQueue) { if (executor == null || completionQueue == null) throw new NullPointerException(); this.executor = executor; this.aes = (executor instanceof AbstractExecutorService) ? (AbstractExecutorService) executor : null; this.completionQueue = completionQueue; } 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; } public Future<V> take() throws InterruptedException { return completionQueue.take(); } public Future<V> poll() { return completionQueue.poll(); } public Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException { return completionQueue.poll(timeout, unit); } }
類 ExecutorCompletionService<V>
所有已實現的介面:
使用提供的 Executor
來執行任務的 CompletionService
。
此類將安排那些完成時提交的任務,把它們放置在可使用 take 訪問的佇列上。
該類非常輕便,適合於在執行幾組任務時臨時使用。
用法示例。 假定您有針對某個問題的一組求解程式,每個求解程式都能返回某種型別的 Result 值,並且您想同時執行它們,使用方法 use(Result r) 處理返回非 null 值的每個求解程式的返回結果。可以這樣編寫程式:
void solve(Executor e, Collection<Callable<Result>> solvers) throws InterruptedException, ExecutionException { CompletionService<Result> ecs = new ExecutorCompletionService<Result>(e); for (Callable<Result> s : solvers) ecs.submit(s); int n = solvers.size(); for (int i = 0; i < n; ++i) { Result r = ecs.take().get(); if (r != null) use(r); } }
假定您想使用任務集中的第一個非 null 結果,而忽略任何遇到異常的任務,並且在第一個任務就緒時取消其他所有任務:
void solve(Executor e, Collection<Callable<Result>> solvers)
throws InterruptedException {
CompletionService<Result> ecs = new ExecutorCompletionService<Result>(e);
int n = solvers.size();
List<Future<Result>> futures = new ArrayList<Future<Result>>(n);
Result result = null;
try {
for (Callable<Result> s : solvers)
futures.add(ecs.submit(s));
for (int i = 0; i < n; ++i) {
try {
Result r = ecs.take().get();
if (r != null) {
result = r;
break;
}
} catch(ExecutionException ignore) {}
}
}
finally {
for (Future<Result> f : futures)
f.cancel(true);
}
if (result != null)
use(result);
}
構造方法摘要
ExecutorCompletionService(Executor executor) 使用為執行基本任務而提供的執行程式建立一個 ExecutorCompletionService,並將 LinkedBlockingQueue 作為完成佇列。 |
ExecutorCompletionService(Executor executor, BlockingQueue<Future<V>> completionQueue) 使用為執行基本任務而提供的執行程式建立一個 ExecutorCompletionService,並將所提供的佇列作為其完成佇列。 |
方法摘要
Future<V> |
poll() 獲取並移除表示下一個已完成任務的 Future,如果不存在這樣的任務,則返回 null。 |
Future<V> |
poll(long timeout, TimeUnit unit) 獲取並移除表示下一個已完成任務的 Future,如果目前不存在這樣的任務,則將等待指定的時間(如果有必要)。 |
Future<V> |
submit(Callable<V> task) 提交要執行的值返回任務,並返回表示掛起的任務結果的 Future。 |
Future<V> |
submit(Runnable task, V result) 提交要執行的 Runnable 任務,並返回一個表示任務完成的 Future,可以提取或輪詢此任務。 |
Future<V> |
take() 獲取並移除表示下一個已完成任務的 Future,如果目前不存在這樣的任務,則等待。 |
從類 java.lang.Object 繼承的方法
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait
ExecutorCompletionService
public ExecutorCompletionService(Executor executor)
使用為執行基本任務而提供的執行程式建立一個 ExecutorCompletionService,並將 LinkedBlockingQueue
作為完成佇列。
引數:
executor
- 要使用的執行程式
丟擲:
NullPointerException
- 如果執行程式為 null
ExecutorCompletionService
public ExecutorCompletionService(Executor executor, BlockingQueue<Future<V>> completionQueue)
使用為執行基本任務而提供的執行程式建立一個 ExecutorCompletionService,並將所提供的佇列作為其完成佇列。
引數:
executor
- 要使用的執行程式
completionQueue
- 用作完成佇列的佇列,通常是專供此服務使用的佇列
丟擲:
NullPointerException
- 如果執行程式或 completionQueue 為 null
submit
public Future<V> submit(Callable<V> task)
從介面 CompletionService
複製的描述
提交要執行的值返回任務,並返回表示掛起的任務結果的 Future。在完成時,可能會提取或輪詢此任務。
指定者:
介面 CompletionService<V>
中的 submit
引數:
task
- 要提交的任務
返回:
一個表示掛起的任務完成的 Future
submit
public Future<V> submit(Runnable task, V result)
從介面 CompletionService
複製的描述
提交要執行的 Runnable 任務,並返回一個表示任務完成的 Future,可以提取或輪詢此任務。
指定者:
介面 CompletionService<V>
中的 submit
引數:
task
- 要提交的任務
result
- 要返回的已成功完成任務的結果
返回:
一個表示掛起的任務完成的 Future,其 get() 方法將返回完成時給出的結果值
take
public Future<V> take() throws InterruptedException
從介面 CompletionService
複製的描述
獲取並移除表示下一個已完成任務的 Future,如果目前不存在這樣的任務,則等待。
指定者:
介面 CompletionService<V>
中的 take
返回:
表示下一個已完成任務的 Future
丟擲:
InterruptedException
- 如果在等待時被中斷
poll
public Future<V> poll()
從介面 CompletionService
複製的描述
獲取並移除表示下一個已完成任務的 Future,如果不存在這樣的任務,則返回 null。
指定者:
介面 CompletionService<V>
中的 poll
返回:
表示下一個已完成任務的 Future;如果不存在這樣的任務,則返回 null
poll
public Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException
從介面 CompletionService
複製的描述
獲取並移除表示下一個已完成任務的 Future,如果目前不存在這樣的任務,則將等待指定的時間(如果有必要)。
指定者:
介面 CompletionService<V>
中的 poll
引數:
timeout
- 放棄之前需要等待的時間長度,以 unit 為時間單位
unit
- 確定如何解釋 timeout 引數的 TimeUnit
返回:
表示下一個已完成任務的 Future;如果等待了指定時間仍然不存在這樣的任務,則返回 null
丟擲:
InterruptedException
- 如果在等待時被中斷