util/concurrent/Executors.java

/*
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain. Use, modify, and
 * redistribute this code in any way without acknowledgement.
 */

package java.util.concurrent;
import java.util.*;

/**
 * Factory and utility methods for the <tt>Executor</tt> classes
 * defined in <tt>java.util.concurrent</tt>.
 *
 * <p>An Executor is a framework for executing Runnables.  The
 * Executor manages queueing and scheduling of tasks, and creation and
 * teardown of threads.  Depending on which concrete Executor class is
 * being used, tasks may execute in a newly created thread, an
 * existing task-execution thread, or the thread calling execute(),
 * and may execute sequentially or concurrently.
 *
 * @since 1.5
 * @see Executor
 * @see ExecutorService
 * @see Future
 *
 * @spec JSR-166
 * @revised $Date: 2003/06/04 15:28:04 $
 * @editor $Author: tim $
 */
public class Executors {

    /**
     * A wrapper class that exposes only the ExecutorService methods
     * of an implementation.
     */
    static private class DelegatedExecutorService implements ExecutorService {
        private final ExecutorService e;
        DelegatedExecutorService(ExecutorService executor) { e = executor; }
        public void execute(Runnable command) { e.execute(command); }
        public void shutdown() { e.shutdown(); }
        public List shutdownNow() { return e.shutdownNow(); }
        public boolean isShutdown() { return e.isShutdown(); }
        public boolean isTerminated() { return e.isTerminated(); }
        public boolean awaitTermination(long timeout, TimeUnit unit)
            throws InterruptedException {
            return e.awaitTermination(timeout, unit);
        }
    }

    /**
     * Creates a thread pool that reuses a fixed set of threads
     * operating off a shared unbounded queue.
     *
     * @param nThreads the number of threads in the pool
     * @return the newly created thread pool
     */
    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new DelegatedExecutorService
            (new ThreadPoolExecutor(nThreads, nThreads,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

    /**
     * Creates a thread pool that reuses a fixed set of threads
     * operating off a shared unbounded queue, using the provided
     * ThreadFactory to create new threads when needed.
     *
     * @param nThreads the number of threads in the pool
     * @param threadfactory the factory to use when creating new threads
     * @return the newly created thread pool
     */
    public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
        return new DelegatedExecutorService
            (new ThreadPoolExecutor(nThreads, nThreads,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>(),
                                    threadFactory, null));
    }

    /**
     * Creates an Executor that uses a single worker thread operating
     * off an unbounded queue. (Note however that if this single
     * thread terminates due to a failure during execution prior to
     * shutdown, a new one will take its place if needed to execute
     * subsequent tasks.)  Tasks are guaranteed to execute
     * sequentially, and no more than one task will be active at any
     * given time.
     *
     * @return the newly-created single-threaded Executor
     */
    public static ExecutorService newSingleThreadExecutor() {
        return new DelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

    /**
     * Creates an Executor that uses a single worker thread operating
     * off an unbounded queue, and uses the provided ThreadFactory to
     * create new threads when needed.
     * @param threadfactory the factory to use when creating new
     * threads
     *
     * @return the newly-created single-threaded Executor
     */
    public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
        return new DelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>(),
                                    threadFactory, null));
    }

    /**
     * Creates a thread pool that creates new threads as needed, but
     * will reuse previously constructed threads when they are
     * available.  These pools will typically improve the performance
     * of programs that execute many short-lived asynchronous tasks.
     * Calls to <tt>execute</tt> will reuse previously constructed
     * threads if available. If no existing thread is available, a new
     * thread will be created and added to the pool. Threads that have
     * not been used for sixty seconds are terminated and removed from
     * the cache. Thus, a pool that remains idle for long enough will
     * not consume any resources.
     *
     * @return the newly created thread pool
     */
    public static ExecutorService newCachedThreadPool() {
        return new DelegatedExecutorService
            (new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                    60, TimeUnit.SECONDS,
                                    new SynchronousQueue<Runnable>()));
    }

    /**
     * Creates a thread pool that creates new threads as needed, but
     * will reuse previously constructed threads when they are
     * available, and uses the provided
     * ThreadFactory to create new threads when needed.
     * @param threadfactory the factory to use when creating new threads
     * @return the newly created thread pool
     */
    public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
        return new DelegatedExecutorService
            (new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                    60, TimeUnit.SECONDS,
                                    new SynchronousQueue<Runnable>(),
                                    threadFactory, null));
    }

    /**
     * Executes a Runnable task and returns a Future representing that
     * task.
     *
     * @param executor the Executor to which the task will be submitted
     * @param task the task to submit
     * @param value the value which will become the return value of
     * the task upon task completion
     * @return a Future representing pending completion of the task
     * @throws ExecutionException if the task cannot be scheduled
     * for execution
     */
    public static <T> Future<T> execute(Executor executor, Runnable task, T value) {
        FutureTask<T> ftask;
        if (executor instanceof ThreadPoolExecutor) {
            ftask = new ThreadPoolFutureTask<T>(
                (ThreadPoolExecutor) executor, task, value);
        } else {
            ftask = new FutureTask<T>(task, value);
        }
        executor.execute(ftask);
        return ftask;
    }

    /**
     * Executes a value-returning task and returns a Future
     * representing the pending results of the task.
     *
     * @param executor the Executor to which the task will be submitted
     * @param task the task to submit
     * @return a Future representing pending completion of the task
     * @throws ExecutionException if task cannot be scheduled for execution
     */
    public static <T> FutureTask<T> execute(Executor executor, Callable<T> task) {
        FutureTask<T> ftask;
        if (executor instanceof ThreadPoolExecutor) {
            ftask = new ThreadPoolFutureTask<T>(
                (ThreadPoolExecutor) executor, task);
        } else {
            ftask = new FutureTask<T>(task);
        }
        executor.execute(ftask);
        return ftask;
    }

    /**
     * Executes a Runnable task and blocks until it completes normally
     * or throws an exception.
     *
     * @param executor the Executor to which the task will be submitted
     * @param task the task to submit
     * @throws ExecutionException if task cannot be scheduled for execution
     */
    public static void invoke(Executor executor, Runnable task)
            throws ExecutionException, InterruptedException {
        FutureTask<Boolean> ftask = new FutureTask(task, Boolean.TRUE);
        executor.execute(ftask);
        ftask.get();
    }

    /**
     * Executes a value-returning task and blocks until it returns a
     * value or throws an exception.
     *
     * @param executor the Executor to which the task will be submitted
     * @param task the task to submit
     * @return a Future representing pending completion of the task
     * @throws ExecutionException if task cannot be scheduled for execution
     */
    public static <T> T invoke(Executor executor, Callable<T> task)
            throws ExecutionException, InterruptedException {
        FutureTask<T> ftask = new FutureTask<T>(task);
        executor.execute(ftask);
        return ftask.get();
    }

    private static class ThreadPoolFutureTask<V> extends FutureTask<V> {

        ThreadPoolFutureTask(ThreadPoolExecutor tpe, Callable<V> callable) {
            super(callable);
            this.tpe = tpe;
        }

        ThreadPoolFutureTask(ThreadPoolExecutor tpe, Runnable runnable, V result) {
            super(runnable, result);
            this.tpe = tpe;
        }

        public boolean cancel(boolean mayInterruptIfRunning) {
            tpe.remove(this); // ignore success/failure
            return super.cancel(mayInterruptIfRunning);
        }

        private final ThreadPoolExecutor tpe;
    }
}
Revision:	1.7
Committed:	Wed Jun 4 15:31:45 2003 UTC (21 years ago) by tim
Branch:	MAIN
Changes since 1.6:	+6 -6 lines
Log Message:	Fixed doc comments to refer to ExecutionException rather than CannotExecuteException.
#	User	Rev	Content
1	tim	1.1	/*
2	dl	1.2	* Written by Doug Lea with assistance from members of JCP JSR-166
3			* Expert Group and released to the public domain. Use, modify, and
4			* redistribute this code in any way without acknowledgement.
5	tim	1.1	*/
6
7			package java.util.concurrent;
8	dl	1.2	import java.util.*;
9	tim	1.1
10			/**
11	dl	1.2	* Factory and utility methods for the <tt>Executor</tt> classes
12			* defined in <tt>java.util.concurrent</tt>.
13	tim	1.1	*
14			* <p>An Executor is a framework for executing Runnables. The
15			* Executor manages queueing and scheduling of tasks, and creation and
16			* teardown of threads. Depending on which concrete Executor class is
17			* being used, tasks may execute in a newly created thread, an
18			* existing task-execution thread, or the thread calling execute(),
19			* and may execute sequentially or concurrently.
20			*
21			* @since 1.5
22			* @see Executor
23	dl	1.2	* @see ExecutorService
24	tim	1.1	* @see Future
25			*
26			* @spec JSR-166
27	tim	1.7	* @revised $Date: 2003/06/04 15:28:04 $
28			* @editor $Author: tim $
29	tim	1.1	*/
30			public class Executors {
31
32			/**
33	dl	1.2	* A wrapper class that exposes only the ExecutorService methods
34			* of an implementation.
35	tim	1.6	*/
36	dl	1.2	static private class DelegatedExecutorService implements ExecutorService {
37			private final ExecutorService e;
38			DelegatedExecutorService(ExecutorService executor) { e = executor; }
39			public void execute(Runnable command) { e.execute(command); }
40			public void shutdown() { e.shutdown(); }
41			public List shutdownNow() { return e.shutdownNow(); }
42			public boolean isShutdown() { return e.isShutdown(); }
43			public boolean isTerminated() { return e.isTerminated(); }
44			public boolean awaitTermination(long timeout, TimeUnit unit)
45			throws InterruptedException {
46			return e.awaitTermination(timeout, unit);
47			}
48			}
49
50			/**
51	tim	1.1	* Creates a thread pool that reuses a fixed set of threads
52			* operating off a shared unbounded queue.
53			*
54			* @param nThreads the number of threads in the pool
55			* @return the newly created thread pool
56			*/
57	dl	1.2	public static ExecutorService newFixedThreadPool(int nThreads) {
58			return new DelegatedExecutorService
59			(new ThreadPoolExecutor(nThreads, nThreads,
60			0L, TimeUnit.MILLISECONDS,
61	dl	1.3	new LinkedBlockingQueue<Runnable>()));
62	dl	1.2	}
63
64			/**
65			* Creates a thread pool that reuses a fixed set of threads
66			* operating off a shared unbounded queue, using the provided
67			* ThreadFactory to create new threads when needed.
68			*
69			* @param nThreads the number of threads in the pool
70			* @param threadfactory the factory to use when creating new threads
71			* @return the newly created thread pool
72			*/
73			public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
74			return new DelegatedExecutorService
75			(new ThreadPoolExecutor(nThreads, nThreads,
76			0L, TimeUnit.MILLISECONDS,
77	dl	1.3	new LinkedBlockingQueue<Runnable>(),
78	dl	1.2	threadFactory, null));
79			}
80
81			/**
82			* Creates an Executor that uses a single worker thread operating
83			* off an unbounded queue. (Note however that if this single
84			* thread terminates due to a failure during execution prior to
85			* shutdown, a new one will take its place if needed to execute
86			* subsequent tasks.) Tasks are guaranteed to execute
87			* sequentially, and no more than one task will be active at any
88			* given time.
89			*
90			* @return the newly-created single-threaded Executor
91			*/
92			public static ExecutorService newSingleThreadExecutor() {
93			return new DelegatedExecutorService
94	tim	1.6	(new ThreadPoolExecutor(1, 1,
95	dl	1.2	0L, TimeUnit.MILLISECONDS,
96	dl	1.3	new LinkedBlockingQueue<Runnable>()));
97	dl	1.2	}
98
99			/**
100			* Creates an Executor that uses a single worker thread operating
101			* off an unbounded queue, and uses the provided ThreadFactory to
102			* create new threads when needed.
103			* @param threadfactory the factory to use when creating new
104			* threads
105			*
106			* @return the newly-created single-threaded Executor
107			*/
108			public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
109			return new DelegatedExecutorService
110	tim	1.6	(new ThreadPoolExecutor(1, 1,
111	dl	1.2	0L, TimeUnit.MILLISECONDS,
112	dl	1.3	new LinkedBlockingQueue<Runnable>(),
113	dl	1.2	threadFactory, null));
114	tim	1.1	}
115
116			/**
117			* Creates a thread pool that creates new threads as needed, but
118			* will reuse previously constructed threads when they are
119			* available. These pools will typically improve the performance
120			* of programs that execute many short-lived asynchronous tasks.
121			* Calls to <tt>execute</tt> will reuse previously constructed
122			* threads if available. If no existing thread is available, a new
123			* thread will be created and added to the pool. Threads that have
124			* not been used for sixty seconds are terminated and removed from
125			* the cache. Thus, a pool that remains idle for long enough will
126			* not consume any resources.
127			*
128			* @return the newly created thread pool
129			*/
130	dl	1.2	public static ExecutorService newCachedThreadPool() {
131			return new DelegatedExecutorService
132			(new ThreadPoolExecutor(0, Integer.MAX_VALUE,
133			60, TimeUnit.SECONDS,
134	dl	1.3	new SynchronousQueue<Runnable>()));
135	tim	1.1	}
136
137			/**
138	dl	1.2	* Creates a thread pool that creates new threads as needed, but
139			* will reuse previously constructed threads when they are
140	tim	1.6	* available, and uses the provided
141	dl	1.2	* ThreadFactory to create new threads when needed.
142			* @param threadfactory the factory to use when creating new threads
143	tim	1.1	* @return the newly created thread pool
144			*/
145	dl	1.2	public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
146			return new DelegatedExecutorService
147			(new ThreadPoolExecutor(0, Integer.MAX_VALUE,
148			60, TimeUnit.SECONDS,
149	dl	1.3	new SynchronousQueue<Runnable>(),
150	dl	1.2	threadFactory, null));
151	tim	1.1	}
152
153			/**
154			* Executes a Runnable task and returns a Future representing that
155			* task.
156			*
157			* @param executor the Executor to which the task will be submitted
158			* @param task the task to submit
159			* @param value the value which will become the return value of
160			* the task upon task completion
161	dl	1.5	* @return a Future representing pending completion of the task
162	tim	1.7	* @throws ExecutionException if the task cannot be scheduled
163	tim	1.1	* for execution
164			*/
165	dl	1.5	public static <T> Future<T> execute(Executor executor, Runnable task, T value) {
166	tim	1.6	FutureTask<T> ftask;
167			if (executor instanceof ThreadPoolExecutor) {
168			ftask = new ThreadPoolFutureTask<T>(
169			(ThreadPoolExecutor) executor, task, value);
170			} else {
171			ftask = new FutureTask<T>(task, value);
172			}
173	tim	1.1	executor.execute(ftask);
174			return ftask;
175			}
176
177			/**
178			* Executes a value-returning task and returns a Future
179			* representing the pending results of the task.
180			*
181			* @param executor the Executor to which the task will be submitted
182			* @param task the task to submit
183			* @return a Future representing pending completion of the task
184	tim	1.7	* @throws ExecutionException if task cannot be scheduled for execution
185	tim	1.1	*/
186	dl	1.4	public static <T> FutureTask<T> execute(Executor executor, Callable<T> task) {
187	tim	1.6	FutureTask<T> ftask;
188			if (executor instanceof ThreadPoolExecutor) {
189			ftask = new ThreadPoolFutureTask<T>(
190			(ThreadPoolExecutor) executor, task);
191			} else {
192			ftask = new FutureTask<T>(task);
193			}
194	tim	1.1	executor.execute(ftask);
195			return ftask;
196			}
197
198			/**
199			* Executes a Runnable task and blocks until it completes normally
200			* or throws an exception.
201			*
202			* @param executor the Executor to which the task will be submitted
203			* @param task the task to submit
204	tim	1.7	* @throws ExecutionException if task cannot be scheduled for execution
205	tim	1.1	*/
206			public static void invoke(Executor executor, Runnable task)
207			throws ExecutionException, InterruptedException {
208	dl	1.4	FutureTask<Boolean> ftask = new FutureTask(task, Boolean.TRUE);
209	tim	1.1	executor.execute(ftask);
210			ftask.get();
211			}
212
213			/**
214			* Executes a value-returning task and blocks until it returns a
215			* value or throws an exception.
216			*
217			* @param executor the Executor to which the task will be submitted
218			* @param task the task to submit
219			* @return a Future representing pending completion of the task
220	tim	1.7	* @throws ExecutionException if task cannot be scheduled for execution
221	tim	1.1	*/
222			public static <T> T invoke(Executor executor, Callable<T> task)
223			throws ExecutionException, InterruptedException {
224			FutureTask<T> ftask = new FutureTask<T>(task);
225			executor.execute(ftask);
226			return ftask.get();
227	tim	1.6	}
228
229			private static class ThreadPoolFutureTask<V> extends FutureTask<V> {
230
231			ThreadPoolFutureTask(ThreadPoolExecutor tpe, Callable<V> callable) {
232			super(callable);
233			this.tpe = tpe;
234			}
235
236			ThreadPoolFutureTask(ThreadPoolExecutor tpe, Runnable runnable, V result) {
237			super(runnable, result);
238			this.tpe = tpe;
239			}
240
241			public boolean cancel(boolean mayInterruptIfRunning) {
242			tpe.remove(this); // ignore success/failure
243			return super.cancel(mayInterruptIfRunning);
244			}
245
246			private final ThreadPoolExecutor tpe;
247	tim	1.1	}
248			}