EDU.oswego.cs.dl.util.concurrent
public class PooledExecutor extends ThreadFactoryUser implements Executor
execute(Runnable command)
, which can be
called instead of directly creating threads to execute commands.
Thread pools can be useful for several, usually intertwined reasons:
When given a choice, this pool always prefers adding a new thread rather than queueing if there are currently fewer than the current getMinimumPoolSize threads running, but otherwise always prefers queuing a request rather than adding a new thread. Thus, if you use an unbounded buffer, you will never have more than getMinimumPoolSize threads running. (Since the default minimumPoolSize is one, you will probably want to explicitly setMinimumPoolSize.)
While queuing can be useful in smoothing out transient bursts of requests, especially in socket-based services, it is not very well behaved when commands continue to arrive on average faster than they can be processed. Using bounds for both the queue and the pool size, along with run-when-blocked policy is often a reasonable response to such possibilities.
Queue sizes and maximum pool sizes can often be traded off for each other. Using large queues and small pools minimizes CPU usage, OS resources, and context-switching overhead, but can lead to artifically low throughput. Especially if tasks frequently block (for example if they are I/O bound), a JVM and underlying OS may be able to schedule time for more threads than you otherwise allow. Use of small queues or queueless handoffs generally requires larger pool sizes, which keeps CPUs busier but may encounter unacceptable scheduling overhead, which also decreases throughput.
execute
request arrives. The default
value is (for all practical purposes) infinite --
Integer.MAX_VALUE
, so should be set in the
constructor or the set method unless you are just using the pool
to minimize construction overhead. Because task handoffs to idle
worker threads require synchronization that in turn relies on JVM
scheduling policies to ensure progress, it is possible that a new
thread will be created even though an existing worker thread has
just become idle but has not progressed to the point at which it
can accept a new task. This phenomenon tends to occur on some
JVMs when bursts of short tasks are executed.
To establish worker threads permanently, use a negative argument to setKeepAliveTime.
execute
requests to
block. There are four supported policies for handling this
problem, and mechanics (based on the Strategy Object pattern) to
allow others in subclasses:
execute
request
runs the task itself. This policy helps guard against lockup.
These cases can never occur if the maximum pool size is unbounded
or the queue is unbounded. In these cases you instead face
potential resource exhaustion.) The execute method does not
throw any checked exceptions in any of these cases since any
errors associated with them must normally be dealt with via
handlers or callbacks. (Although in some cases, these might be
associated with throwing unchecked exceptions.) You may wish to
add special implementations even if you choose one of the listed
policies. For example, the supplied Discard policy does not
inform the caller of the drop. You could add your own version
that does so. Since choice of policies is normally a system-wide
decision, selecting a policy affects all calls to
execute
. If for some reason you would instead like
to make per-call decisions, you could add variant versions of the
execute
method (for example,
executeIfWouldNotBlock
) in subclasses.
Also, if you are using some form of queuing, you may wish to call method drain() to remove (and return) unprocessed commands from the queue after shutting down the pool and its clients. If you need to be sure these commands are processed, you can then run() each of the commands in the list returned by drain().
Usage examples.
Probably the most common use of pools is in statics or singletons accessible from a number of classes in a package; for example:
class MyPool { // initialize to use a maximum of 8 threads. static PooledExecutor pool = new PooledExecutor(8); }Here are some sample variants in initialization:
pool = new PooledExecutor(new BoundedBuffer(10), 100); pool.setMinimumPoolSize(4);
pool = new PooledExecutor(new BoundedBuffer(10), 100); pool.setMinimumPoolSize(4); pool.setKeepAliveTime(1000 * 60 * 5); pool.createThreads(9);
pool = new PooledExecutor(new BoundedBuffer(10), 100); pool.setMinimumPoolSize(4); pool.setKeepAliveTime(1000 * 60 * 5); pool.abortWhenBlocked(); pool.createThreads(9);
pool = new PooledExecutor(new LinkedQueue()); pool.setKeepAliveTime(-1); // live forever pool.createThreads(5);
Usage notes.
Pools do not mesh well with using thread-specific storage via java.lang.ThreadLocal. ThreadLocal relies on the identity of a thread executing a particular task. Pools use the same thread to perform different tasks.
If you need a policy not handled by the parameters in this class consider writing a subclass.
Version note: Previous versions of this class relied on ThreadGroups for aggregate control. This has been removed, and the method interruptAll added, to avoid differences in behavior across JVMs.
Nested Class Summary | |
---|---|
protected class | PooledExecutor.AbortWhenBlocked Class defining Abort action. |
interface | PooledExecutor.BlockedExecutionHandler
Class for actions to take when execute() blocks. |
protected class | PooledExecutor.DiscardOldestWhenBlocked
Class defining DiscardOldest action. |
protected class | PooledExecutor.DiscardWhenBlocked Class defining Discard action. |
protected class | PooledExecutor.RunWhenBlocked Class defining Run action. |
protected class | PooledExecutor.WaitWhenBlocked Class defining Wait action. |
protected class | PooledExecutor.Worker
Class defining the basic run loop for pooled threads.
|
Field Summary | |
---|---|
protected PooledExecutor.BlockedExecutionHandler | blockedExecutionHandler_ The current handler for unserviceable requests. |
static long | DEFAULT_KEEPALIVETIME
The maximum time to keep worker threads alive waiting for new
tasks; used if not otherwise specified. |
static int | DEFAULT_MAXIMUMPOOLSIZE
The maximum pool size; used if not otherwise specified. |
static int | DEFAULT_MINIMUMPOOLSIZE
The minimum pool size; used if not otherwise specified. |
protected Channel | handOff_
The channel used to hand off the command to a thread in the pool.
|
protected long | keepAliveTime_ The maximum time for an idle thread to wait for new task. |
protected int | maximumPoolSize_ The maximum number of threads allowed in pool. |
protected int | minimumPoolSize_ The minumum number of threads to maintain in pool. |
protected int | poolSize_ Current pool size. |
protected boolean | shutdown_
Shutdown flag - latches true when a shutdown method is called
in order to disable queuing/handoffs of new tasks.
|
protected Map | threads_
The set of active threads, declared as a map from workers to
their threads. |
Constructor Summary | |
---|---|
PooledExecutor()
Create a new pool with all default settings
| |
PooledExecutor(int maxPoolSize)
Create a new pool with all default settings except
for maximum pool size.
| |
PooledExecutor(Channel channel)
Create a new pool that uses the supplied Channel for queuing, and
with all default parameter settings.
| |
PooledExecutor(Channel channel, int maxPoolSize)
Create a new pool that uses the supplied Channel for queuing, and
with all default parameter settings except for maximum pool size.
|
Method Summary | |
---|---|
void | abortWhenBlocked()
Set the policy for blocked execution to be to
throw a RuntimeException.
|
protected void | addThread(Runnable command)
Create and start a thread to handle a new command. |
boolean | awaitTerminationAfterShutdown(long maxWaitTime)
Wait for a shutdown pool to fully terminate, or until the timeout
has expired. |
void | awaitTerminationAfterShutdown()
Wait for a shutdown pool to fully terminate. |
int | createThreads(int numberOfThreads)
Create and start up to numberOfThreads threads in the pool.
|
void | discardOldestWhenBlocked()
Set the policy for blocked execution to be to discard the oldest
unhandled request
|
void | discardWhenBlocked()
Set the policy for blocked execution to be to return without
executing the request.
|
List | drain()
Remove all unprocessed tasks from pool queue, and return them in
a java.util.List. |
void | execute(Runnable command)
Arrange for the given command to be executed by a thread in this
pool. |
PooledExecutor.BlockedExecutionHandler | getBlockedExecutionHandler() Get the handler for blocked execution * |
long | getKeepAliveTime()
Return the number of milliseconds to keep threads alive waiting
for new commands. |
int | getMaximumPoolSize()
Return the maximum number of threads to simultaneously execute
New unqueued requests will be handled according to the current
blocking policy once this limit is exceeded.
|
int | getMinimumPoolSize()
Return the minimum number of threads to simultaneously execute.
|
int | getPoolSize()
Return the current number of active threads in the pool. |
protected Runnable | getTask()
Get a task from the handoff queue, or null if shutting down.
|
void | interruptAll()
Interrupt all threads in the pool, causing them all to
terminate. |
boolean | isTerminatedAfterShutdown()
Return true if a shutDown method has succeeded in terminating all
threads. |
void | runWhenBlocked()
Set the policy for blocked execution to be that the current
thread executes the command if there are no available threads in
the pool.
|
void | setBlockedExecutionHandler(PooledExecutor.BlockedExecutionHandler h) Set the handler for blocked execution * |
void | setKeepAliveTime(long msecs)
Set the number of milliseconds to keep threads alive waiting for
new commands. |
void | setMaximumPoolSize(int newMaximum)
Set the maximum number of threads to use. |
void | setMinimumPoolSize(int newMinimum)
Set the minimum number of threads to use. |
void | shutdownAfterProcessingCurrentlyQueuedTasks()
Terminate threads after processing all elements currently in
queue. |
void | shutdownAfterProcessingCurrentlyQueuedTasks(PooledExecutor.BlockedExecutionHandler handler)
Terminate threads after processing all elements currently in
queue. |
void | shutdownNow()
Interrupt all threads and disable construction of new
threads. |
void | shutdownNow(PooledExecutor.BlockedExecutionHandler handler)
Interrupt all threads and disable construction of new
threads. |
void | waitWhenBlocked()
Set the policy for blocked execution to be to wait until a thread
is available, unless the pool has been shut down, in which case
the action is discarded.
|
protected void | workerDone(PooledExecutor.Worker w)
Cleanup method called upon termination of worker thread.
|
Parameters: maxWaitTime the maximum time in milliseconds to wait
Returns: true if the pool has terminated within the max wait period
Throws: IllegalStateException if shutdown has not been requested InterruptedException if the current thread has been interrupted in the course of waiting
Throws: IllegalStateException if shutdown has not been requested InterruptedException if the current thread has been interrupted in the course of waiting
List tasks = pool.drain(); for (Iterator it = tasks.iterator(); it.hasNext();) ( (Runnable)(it.next()) ).run();
Throws: IllegalArgumentException if less or equal to zero. (It is not considered an error to set the maximum to be less than than the minimum. However, in this case there are no guarantees about behavior.)
Throws: IllegalArgumentException if less than zero. (It is not considered an error to set the minimum to be greater than the maximum. However, in this case there are no guarantees about behavior.)