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