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/04 15:31:45 $
 * @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.8
Committed:	Fri Jun 6 18:42:17 2003 UTC (21 years ago) by dl
Branch:	MAIN
Changes since 1.7:	+1 -8 lines
Log Message:	Added to emulation Fixed some javadoc format errors
#	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
10	/**
11	* Factory and utility methods for the <tt>Executor</tt> classes
12	* defined in <tt>java.util.concurrent</tt>.
13	*
14	* @since 1.5
15	* @see Executor
16	* @see ExecutorService
17	* @see Future
18	*
19	* @spec JSR-166
20	* @revised $Date: 2003/06/04 15:31:45 $
21	* @editor $Author: tim $
22	*/
23	public class Executors {
24
25	/**
26	* A wrapper class that exposes only the ExecutorService methods
27	* of an implementation.
28	*/
29	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	* 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	public static ExecutorService newFixedThreadPool(int nThreads) {
51	return new DelegatedExecutorService
52	(new ThreadPoolExecutor(nThreads, nThreads,
53	0L, TimeUnit.MILLISECONDS,
54	new LinkedBlockingQueue<Runnable>()));
55	}
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	new LinkedBlockingQueue<Runnable>(),
71	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	(new ThreadPoolExecutor(1, 1,
88	0L, TimeUnit.MILLISECONDS,
89	new LinkedBlockingQueue<Runnable>()));
90	}
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	(new ThreadPoolExecutor(1, 1,
104	0L, TimeUnit.MILLISECONDS,
105	new LinkedBlockingQueue<Runnable>(),
106	threadFactory, null));
107	}
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	public static ExecutorService newCachedThreadPool() {
124	return new DelegatedExecutorService
125	(new ThreadPoolExecutor(0, Integer.MAX_VALUE,
126	60, TimeUnit.SECONDS,
127	new SynchronousQueue<Runnable>()));
128	}
129
130	/**
131	* Creates a thread pool that creates new threads as needed, but
132	* will reuse previously constructed threads when they are
133	* available, and uses the provided
134	* ThreadFactory to create new threads when needed.
135	* @param threadfactory the factory to use when creating new threads
136	* @return the newly created thread pool
137	*/
138	public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
139	return new DelegatedExecutorService
140	(new ThreadPoolExecutor(0, Integer.MAX_VALUE,
141	60, TimeUnit.SECONDS,
142	new SynchronousQueue<Runnable>(),
143	threadFactory, null));
144	}
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	* @return a Future representing pending completion of the task
155	* @throws ExecutionException if the task cannot be scheduled
156	* for execution
157	*/
158	public static <T> Future<T> execute(Executor executor, Runnable task, T value) {
159	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	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	* @throws ExecutionException if task cannot be scheduled for execution
178	*/
179	public static <T> FutureTask<T> execute(Executor executor, Callable<T> task) {
180	FutureTask<T> ftask;
181	if (executor instanceof ThreadPoolExecutor) {
182	ftask = new ThreadPoolFutureTask<T>(
183	(ThreadPoolExecutor) executor, task);
184	} else {
185	ftask = new FutureTask<T>(task);
186	}
187	executor.execute(ftask);
188	return ftask;
189	}
190
191	/**
192	* Executes a Runnable task and blocks until it completes normally
193	* or throws an exception.
194	*
195	* @param executor the Executor to which the task will be submitted
196	* @param task the task to submit
197	* @throws ExecutionException if task cannot be scheduled for execution
198	*/
199	public static void invoke(Executor executor, Runnable task)
200	throws ExecutionException, InterruptedException {
201	FutureTask<Boolean> ftask = new FutureTask(task, Boolean.TRUE);
202	executor.execute(ftask);
203	ftask.get();
204	}
205
206	/**
207	* Executes a value-returning task and blocks until it returns a
208	* value or throws an exception.
209	*
210	* @param executor the Executor to which the task will be submitted
211	* @param task the task to submit
212	* @return a Future representing pending completion of the task
213	* @throws ExecutionException if task cannot be scheduled for execution
214	*/
215	public static <T> T invoke(Executor executor, Callable<T> task)
216	throws ExecutionException, InterruptedException {
217	FutureTask<T> ftask = new FutureTask<T>(task);
218	executor.execute(ftask);
219	return ftask.get();
220	}
221
222	private static class ThreadPoolFutureTask<V> extends FutureTask<V> {
223
224	ThreadPoolFutureTask(ThreadPoolExecutor tpe, Callable<V> callable) {
225	super(callable);
226	this.tpe = tpe;
227	}
228
229	ThreadPoolFutureTask(ThreadPoolExecutor tpe, Runnable runnable, V result) {
230	super(runnable, result);
231	this.tpe = tpe;
232	}
233
234	public boolean cancel(boolean mayInterruptIfRunning) {
235	tpe.remove(this); // ignore success/failure
236	return super.cancel(mayInterruptIfRunning);
237	}
238
239	private final ThreadPoolExecutor tpe;
240	}
241	}