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/07/31 19:49:42 $
 * @editor $Author: tim $
 * @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.
     *
     * @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.
     *
     * @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.
     *
     * @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 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> FutureTask<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
     */
    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
     */
    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();
    }

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