1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175
/* * @(#)ExecutorCompletionService.java 1.4 06/01/30 * * Copyright 2006 Sun Microsystems, Inc. All rights reserved. * SUN PROPRIETARY/CONFIDENTIAL. Use is subject to license terms. */ package java.util.concurrent; /** * A {@link CompletionService} that uses a supplied {@link Executor} * to execute tasks. This class arranges that submitted tasks are, * upon completion, placed on a queue accessible using <tt>take</tt>. * The class is lightweight enough to be suitable for transient use * when processing groups of tasks. * * <p> * * <b>Usage Examples.</b> * * Suppose you have a set of solvers for a certain problem, each * returning a value of some type <tt>Result</tt>, and would like to * run them concurrently, processing the results of each of them that * return a non-null value, in some method <tt>use(Result r)</tt>. You * could write this as: * * <pre> * 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); * } * } * </pre> * * Suppose instead that you would like to use the first non-null result * of the set of tasks, ignoring any that encounter exceptions, * and cancelling all other tasks when the first one is ready: * * <pre> * 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); * } * </pre> */ public class ExecutorCompletionService<V> implements CompletionService<V> { private final Executor executor; private final AbstractExecutorService aes; private final BlockingQueue<Future<V>> completionQueue; /** * 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; } 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); } /** * Creates an ExecutorCompletionService using the supplied * executor for base task execution and a * {@link LinkedBlockingQueue} as a completion queue. * * @param executor the executor to use * @throws NullPointerException if executor is <tt>null</tt> */ 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>>(); } /** * Creates an ExecutorCompletionService using the supplied * executor for base task execution and the supplied queue as its * completion queue. * * @param executor the executor to use * @param completionQueue the queue to use as the completion queue * normally one dedicated for use by this service * @throws NullPointerException if executor or completionQueue are <tt>null</tt> */ 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); } }