|
||||||||
PREV CLASS NEXT CLASS | FRAMES NO FRAMES | |||||||
SUMMARY: NESTED | FIELD | CONSTR | METHOD | DETAIL: FIELD | CONSTR | METHOD |
java.lang.Object EDU.oswego.cs.dl.util.concurrent.ThreadFactoryUser EDU.oswego.cs.dl.util.concurrent.PooledExecutor
public class PooledExecutor
A tunable, extensible thread pool class. The main supported public
method is 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.
[ Introduction to this package. ]
Nested Class Summary | |
---|---|
protected class |
PooledExecutor.AbortWhenBlocked
Class defining Abort action. |
static 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. |
Nested classes/interfaces inherited from class EDU.oswego.cs.dl.util.concurrent.ThreadFactoryUser |
---|
ThreadFactoryUser.DefaultThreadFactory |
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 java.util.Map |
threads_
The set of active threads, declared as a map from workers to their threads. |
Fields inherited from class EDU.oswego.cs.dl.util.concurrent.ThreadFactoryUser |
---|
threadFactory_ |
Constructor Summary | |
---|---|
PooledExecutor()
Create a new pool with all default settings |
|
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. |
|
PooledExecutor(int maxPoolSize)
Create a new pool with all default 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(java.lang.Runnable command)
Create and start a thread to handle a new command. |
void |
awaitTerminationAfterShutdown()
Wait for a shutdown pool to fully terminate. |
boolean |
awaitTerminationAfterShutdown(long maxWaitTime)
Wait for a shutdown pool to fully terminate, or until the timeout has expired. |
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. |
java.util.List |
drain()
Remove all unprocessed tasks from pool queue, and return them in a java.util.List. |
void |
execute(java.lang.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 java.lang.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. |
Methods inherited from class EDU.oswego.cs.dl.util.concurrent.ThreadFactoryUser |
---|
getThreadFactory, setThreadFactory |
Methods inherited from class java.lang.Object |
---|
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait |
Field Detail |
---|
public static final int DEFAULT_MAXIMUMPOOLSIZE
public static final int DEFAULT_MINIMUMPOOLSIZE
public static final long DEFAULT_KEEPALIVETIME
protected int maximumPoolSize_
protected int minimumPoolSize_
protected int poolSize_
protected long keepAliveTime_
protected boolean shutdown_
protected final Channel handOff_
protected final java.util.Map threads_
protected PooledExecutor.BlockedExecutionHandler blockedExecutionHandler_
Constructor Detail |
---|
public PooledExecutor()
public PooledExecutor(int maxPoolSize)
public PooledExecutor(Channel channel)
public PooledExecutor(Channel channel, int maxPoolSize)
Method Detail |
---|
public int getMaximumPoolSize()
public void setMaximumPoolSize(int newMaximum)
java.lang.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.)public int getMinimumPoolSize()
public void setMinimumPoolSize(int newMinimum)
java.lang.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.)public int getPoolSize()
public long getKeepAliveTime()
public void setKeepAliveTime(long msecs)
public PooledExecutor.BlockedExecutionHandler getBlockedExecutionHandler()
public void setBlockedExecutionHandler(PooledExecutor.BlockedExecutionHandler h)
protected void addThread(java.lang.Runnable command)
public int createThreads(int numberOfThreads)
public void interruptAll()
public void shutdownNow()
public void shutdownNow(PooledExecutor.BlockedExecutionHandler handler)
public void shutdownAfterProcessingCurrentlyQueuedTasks()
public void shutdownAfterProcessingCurrentlyQueuedTasks(PooledExecutor.BlockedExecutionHandler handler)
public boolean isTerminatedAfterShutdown()
public boolean awaitTerminationAfterShutdown(long maxWaitTime) throws java.lang.InterruptedException
maxWaitTime
- the maximum time in milliseconds to wait
java.lang.IllegalStateException
- if shutdown has not been requested
java.lang.InterruptedException
- if the current thread has been interrupted in the course of waitingpublic void awaitTerminationAfterShutdown() throws java.lang.InterruptedException
java.lang.IllegalStateException
- if shutdown has not been requested
java.lang.InterruptedException
- if the current thread has been interrupted in the course of waitingpublic java.util.List drain()
List tasks = pool.drain(); for (Iterator it = tasks.iterator(); it.hasNext();) ( (Runnable)(it.next()) ).run();
protected void workerDone(PooledExecutor.Worker w)
protected java.lang.Runnable getTask() throws java.lang.InterruptedException
java.lang.InterruptedException
public void runWhenBlocked()
public void waitWhenBlocked()
public void discardWhenBlocked()
public void abortWhenBlocked()
public void discardOldestWhenBlocked()
public void execute(java.lang.Runnable command) throws java.lang.InterruptedException
execute
in interface Executor
java.lang.InterruptedException
|
||||||||
PREV CLASS NEXT CLASS | FRAMES NO FRAMES | |||||||
SUMMARY: NESTED | FIELD | CONSTR | METHOD | DETAIL: FIELD | CONSTR | METHOD |