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.*;
import java.security.AccessControlContext;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.security.PrivilegedExceptionAction;

/**
 * Factory and utility methods for {@link Executor}, {@link
 * ExecutorService}, {@link Future}, and {@link Cancellable} classes
 * defined in this package.
 *
 * @since 1.5
 * @author Doug Lea
 */
public class Executors {

    /**
     * A wrapper class that exposes only the ExecutorService methods
     * of an implementation.
     */
    private static 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. If any thread
     * terminates due to a failure during execution prior to shutdown,
     * a new one will take its place if needed to execute subsequent
     * tasks.
     *
     * @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));
    }

    /**
     * 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. This method is equivalent in effect to
     *<tt>new FixedThreadPool(1)</tt>.
     *
     * @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));
    }

    /**
     * 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. Note that pools with similar
     * properties but different details (for example, timeout parameters)
     * may be created using {@link ThreadPoolExecutor} constructors.
     *
     * @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));
    }

    /**
     * Executes a Runnable task and returns a Cancellable representing that
     * task.
     *
     * @param executor the Executor to which the task will be submitted
     * @param task the task to submit
     * @return a Cancellable representing pending completion of the task
     * @throws RejectedExecutionException if task cannot be scheduled
     * for execution
     */
    public static Cancellable execute(Executor executor, Runnable task) {
        FutureTask<Boolean> ftask = new FutureTask<Boolean>(task, Boolean.TRUE);
        executor.execute(ftask);
        return ftask;
    }

    /**
     * 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 = 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> Future<T> execute(Executor executor, Callable<T> task) {
        FutureTask<T> 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
     * @throws ExecutionException if the task encountered an exception
     * while executing
     */
    public static void invoke(Executor executor, Runnable task)
            throws ExecutionException, InterruptedException {
        FutureTask<Boolean> ftask = new FutureTask<Boolean>(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
     * @throws InterruptedException if interrupted while waiting for
     * completion
     * @throws ExecutionException if the task encountered an exception
     * while executing
     */
    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();
    }


    /**
     * Executes a privileged action under the current access control 
     * context and returns a Future representing the pending result 
     * object of that action.
     *
     * @param executor the Executor to which the task will be submitted
     * @param action the action to submit
     * @return a Future representing pending completion of the action
     * @throws RejectedExecutionException if action cannot be scheduled
     * for execution
     */
    public static Future<Object> execute(Executor executor, PrivilegedAction action) {
        Callable<Object> task = new PrivilegedActionAdapter(action);
        FutureTask<Object> future = new PrivilegedFutureTask<Object>(task);
        executor.execute(future);
        return future;
    }

    /**
     * Executes a privileged exception action under the current access control 
     * context and returns a Future representing the pending result 
     * object of that action.
     *
     * @param executor the Executor to which the task will be submitted
     * @param action the action to submit
     * @return a Future representing pending completion of the action
     * @throws RejectedExecutionException if action cannot be scheduled
     * for execution
     */
    public static Future<Object> execute(Executor executor, PrivilegedExceptionAction action) {
        Callable<Object> task = new PrivilegedExceptionActionAdapter(action);
        FutureTask<Object> future = new PrivilegedFutureTask<Object>(task);
        executor.execute(future);
        return future;
    }
    

    private static class PrivilegedActionAdapter implements Callable<Object> {
        PrivilegedActionAdapter(PrivilegedAction action) {
            this.action = action;
        }
        public Object call () {
            return action.run();
        }
        private final PrivilegedAction action;
    }
    
    private static class PrivilegedExceptionActionAdapter implements Callable<Object> {
        PrivilegedExceptionActionAdapter(PrivilegedExceptionAction action) {
            this.action = action;
        }
        public Object call () throws Exception {
            return action.run();
        }
        private final PrivilegedExceptionAction action;
    }
        
        
    /** Cannot instantiate. */
    private Executors() {}
}
Revision:	1.21
Committed:	Tue Oct 28 13:25:01 2003 UTC (20 years, 7 months ago) by tim
Branch:	MAIN
Changes since 1.20:	+2 -36 lines
Log Message:	Remove explicit CCL/ACC constructors and class methods
#	Content
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	*/
6
7	package java.util.concurrent;
8	import java.util.*;
9	import java.security.AccessControlContext;
10	import java.security.AccessController;
11	import java.security.PrivilegedAction;
12	import java.security.PrivilegedExceptionAction;
13
14	/**
15	* Factory and utility methods for {@link Executor}, {@link
16	* ExecutorService}, {@link Future}, and {@link Cancellable} classes
17	* defined in this package.
18	*
19	* @since 1.5
20	* @author Doug Lea
21	*/
22	public class Executors {
23
24	/**
25	* A wrapper class that exposes only the ExecutorService methods
26	* of an implementation.
27	*/
28	private static class DelegatedExecutorService implements ExecutorService {
29	private final ExecutorService e;
30	DelegatedExecutorService(ExecutorService executor) { e = executor; }
31	public void execute(Runnable command) { e.execute(command); }
32	public void shutdown() { e.shutdown(); }
33	public List shutdownNow() { return e.shutdownNow(); }
34	public boolean isShutdown() { return e.isShutdown(); }
35	public boolean isTerminated() { return e.isTerminated(); }
36	public boolean awaitTermination(long timeout, TimeUnit unit)
37	throws InterruptedException {
38	return e.awaitTermination(timeout, unit);
39	}
40	}
41
42	/**
43	* Creates a thread pool that reuses a fixed set of threads
44	* operating off a shared unbounded queue. If any thread
45	* terminates due to a failure during execution prior to shutdown,
46	* a new one will take its place if needed to execute subsequent
47	* tasks.
48	*
49	* @param nThreads the number of threads in the pool
50	* @return the newly created thread pool
51	*/
52	public static ExecutorService newFixedThreadPool(int nThreads) {
53	return new DelegatedExecutorService
54	(new ThreadPoolExecutor(nThreads, nThreads,
55	0L, TimeUnit.MILLISECONDS,
56	new LinkedBlockingQueue<Runnable>()));
57	}
58
59	/**
60	* Creates a thread pool that reuses a fixed set of threads
61	* operating off a shared unbounded queue, using the provided
62	* ThreadFactory to create new threads when needed.
63	*
64	* @param nThreads the number of threads in the pool
65	* @param threadFactory the factory to use when creating new threads
66	* @return the newly created thread pool
67	*/
68	public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
69	return new DelegatedExecutorService
70	(new ThreadPoolExecutor(nThreads, nThreads,
71	0L, TimeUnit.MILLISECONDS,
72	new LinkedBlockingQueue<Runnable>(),
73	threadFactory));
74	}
75
76	/**
77	* Creates an Executor that uses a single worker thread operating
78	* off an unbounded queue. (Note however that if this single
79	* thread terminates due to a failure during execution prior to
80	* shutdown, a new one will take its place if needed to execute
81	* subsequent tasks.) Tasks are guaranteed to execute
82	* sequentially, and no more than one task will be active at any
83	* given time. This method is equivalent in effect to
84	*<tt>new FixedThreadPool(1)</tt>.
85	*
86	* @return the newly-created single-threaded Executor
87	*/
88	public static ExecutorService newSingleThreadExecutor() {
89	return new DelegatedExecutorService
90	(new ThreadPoolExecutor(1, 1,
91	0L, TimeUnit.MILLISECONDS,
92	new LinkedBlockingQueue<Runnable>()));
93	}
94
95	/**
96	* Creates an Executor that uses a single worker thread operating
97	* off an unbounded queue, and uses the provided ThreadFactory to
98	* create new threads when needed.
99	* @param threadFactory the factory to use when creating new
100	* threads
101	*
102	* @return the newly-created single-threaded Executor
103	*/
104	public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
105	return new DelegatedExecutorService
106	(new ThreadPoolExecutor(1, 1,
107	0L, TimeUnit.MILLISECONDS,
108	new LinkedBlockingQueue<Runnable>(),
109	threadFactory));
110	}
111
112	/**
113	* Creates a thread pool that creates new threads as needed, but
114	* will reuse previously constructed threads when they are
115	* available. These pools will typically improve the performance
116	* of programs that execute many short-lived asynchronous tasks.
117	* Calls to <tt>execute</tt> will reuse previously constructed
118	* threads if available. If no existing thread is available, a new
119	* thread will be created and added to the pool. Threads that have
120	* not been used for sixty seconds are terminated and removed from
121	* the cache. Thus, a pool that remains idle for long enough will
122	* not consume any resources. Note that pools with similar
123	* properties but different details (for example, timeout parameters)
124	* may be created using {@link ThreadPoolExecutor} constructors.
125	*
126	* @return the newly created thread pool
127	*/
128	public static ExecutorService newCachedThreadPool() {
129	return new DelegatedExecutorService
130	(new ThreadPoolExecutor(0, Integer.MAX_VALUE,
131	60, TimeUnit.SECONDS,
132	new SynchronousQueue<Runnable>()));
133	}
134
135	/**
136	* Creates a thread pool that creates new threads as needed, but
137	* will reuse previously constructed threads when they are
138	* available, and uses the provided
139	* ThreadFactory to create new threads when needed.
140	* @param threadFactory the factory to use when creating new threads
141	* @return the newly created thread pool
142	*/
143	public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
144	return new DelegatedExecutorService
145	(new ThreadPoolExecutor(0, Integer.MAX_VALUE,
146	60, TimeUnit.SECONDS,
147	new SynchronousQueue<Runnable>(),
148	threadFactory));
149	}
150
151	/**
152	* Executes a Runnable task and returns a Cancellable representing that
153	* task.
154	*
155	* @param executor the Executor to which the task will be submitted
156	* @param task the task to submit
157	* @return a Cancellable representing pending completion of the task
158	* @throws RejectedExecutionException if task cannot be scheduled
159	* for execution
160	*/
161	public static Cancellable execute(Executor executor, Runnable task) {
162	FutureTask<Boolean> ftask = new FutureTask<Boolean>(task, Boolean.TRUE);
163	executor.execute(ftask);
164	return ftask;
165	}
166
167	/**
168	* Executes a Runnable task and returns a Future representing that
169	* task.
170	*
171	* @param executor the Executor to which the task will be submitted
172	* @param task the task to submit
173	* @param value the value which will become the return value of
174	* the task upon task completion
175	* @return a Future representing pending completion of the task
176	* @throws RejectedExecutionException if task cannot be scheduled
177	* for execution
178	*/
179	public static <T> Future<T> execute(Executor executor, Runnable task, T value) {
180	FutureTask<T> ftask = new FutureTask<T>(task, value);
181	executor.execute(ftask);
182	return ftask;
183	}
184
185	/**
186	* Executes a value-returning task and returns a Future
187	* representing the pending results of the task.
188	*
189	* @param executor the Executor to which the task will be submitted
190	* @param task the task to submit
191	* @return a Future representing pending completion of the task
192	* @throws RejectedExecutionException if task cannot be scheduled
193	* for execution
194	*/
195	public static <T> Future<T> execute(Executor executor, Callable<T> task) {
196	FutureTask<T> ftask = new FutureTask<T>(task);
197	executor.execute(ftask);
198	return ftask;
199	}
200
201	/**
202	* Executes a Runnable task and blocks until it completes normally
203	* or throws an exception.
204	*
205	* @param executor the Executor to which the task will be submitted
206	* @param task the task to submit
207	* @throws RejectedExecutionException if task cannot be scheduled
208	* for execution
209	* @throws ExecutionException if the task encountered an exception
210	* while executing
211	*/
212	public static void invoke(Executor executor, Runnable task)
213	throws ExecutionException, InterruptedException {
214	FutureTask<Boolean> ftask = new FutureTask<Boolean>(task, Boolean.TRUE);
215	executor.execute(ftask);
216	ftask.get();
217	}
218
219	/**
220	* Executes a value-returning task and blocks until it returns a
221	* value or throws an exception.
222	*
223	* @param executor the Executor to which the task will be submitted
224	* @param task the task to submit
225	* @return a Future representing pending completion of the task
226	* @throws RejectedExecutionException if task cannot be scheduled
227	* for execution
228	* @throws InterruptedException if interrupted while waiting for
229	* completion
230	* @throws ExecutionException if the task encountered an exception
231	* while executing
232	*/
233	public static <T> T invoke(Executor executor, Callable<T> task)
234	throws ExecutionException, InterruptedException {
235	FutureTask<T> ftask = new FutureTask<T>(task);
236	executor.execute(ftask);
237	return ftask.get();
238	}
239
240
241	/**
242	* Executes a privileged action under the current access control
243	* context and returns a Future representing the pending result
244	* object of that action.
245	*
246	* @param executor the Executor to which the task will be submitted
247	* @param action the action to submit
248	* @return a Future representing pending completion of the action
249	* @throws RejectedExecutionException if action cannot be scheduled
250	* for execution
251	*/
252	public static Future<Object> execute(Executor executor, PrivilegedAction action) {
253	Callable<Object> task = new PrivilegedActionAdapter(action);
254	FutureTask<Object> future = new PrivilegedFutureTask<Object>(task);
255	executor.execute(future);
256	return future;
257	}
258
259	/**
260	* Executes a privileged exception action under the current access control
261	* context and returns a Future representing the pending result
262	* object of that action.
263	*
264	* @param executor the Executor to which the task will be submitted
265	* @param action the action to submit
266	* @return a Future representing pending completion of the action
267	* @throws RejectedExecutionException if action cannot be scheduled
268	* for execution
269	*/
270	public static Future<Object> execute(Executor executor, PrivilegedExceptionAction action) {
271	Callable<Object> task = new PrivilegedExceptionActionAdapter(action);
272	FutureTask<Object> future = new PrivilegedFutureTask<Object>(task);
273	executor.execute(future);
274	return future;
275	}
276
277
278	private static class PrivilegedActionAdapter implements Callable<Object> {
279	PrivilegedActionAdapter(PrivilegedAction action) {
280	this.action = action;
281	}
282	public Object call () {
283	return action.run();
284	}
285	private final PrivilegedAction action;
286	}
287
288	private static class PrivilegedExceptionActionAdapter implements Callable<Object> {
289	PrivilegedExceptionActionAdapter(PrivilegedExceptionAction action) {
290	this.action = action;
291	}
292	public Object call () throws Exception {
293	return action.run();
294	}
295	private final PrivilegedExceptionAction action;
296	}
297
298
299	/** Cannot instantiate. */
300	private Executors() {}
301	}