|
Colt 1.0.2 | |||||||||
PREV CLASS NEXT CLASS | FRAMES NO FRAMES | |||||||||
SUMMARY: INNER | FIELD | CONSTR | METHOD | DETAIL: FIELD | CONSTR | METHOD |
java.lang.Object | +--edu.oswego.cs.dl.util.concurrent.ThreadFactoryUser | +--edu.oswego.cs.dl.util.concurrent.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.
(Again, 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.
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.waitWhenBlocked(); 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. ]
Field Summary | |
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. |
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 | |
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. |
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 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. |
void |
interruptAll()
Interrupt all threads in the pool, causing them all to terminate. |
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 |
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 (via interruptAll) threads after processing all elements currently in queue. |
void |
waitWhenBlocked()
Set the policy for blocked execution to be to wait until a thread is available. |
Methods inherited from class edu.oswego.cs.dl.util.concurrent.ThreadFactoryUser |
getThreadFactory, setThreadFactory |
Methods inherited from class java.lang.Object |
equals, 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
Constructor Detail |
public PooledExecutor()
public PooledExecutor(int maxPoolSize)
public PooledExecutor(Channel channel)
public PooledExecutor(Channel channel, int maxPoolSize)
Method Detail |
public int createThreads(int numberOfThreads)
public void discardOldestWhenBlocked()
public void discardWhenBlocked()
public List drain()
List tasks = pool.drain(); for (Iterator it = tasks.iterator(); it.hasNext();) ( (Runnable)(it.next()) ).run();
public void execute(Runnable command) throws InterruptedException
execute
in interface Executor
public long getKeepAliveTime()
public int getMaximumPoolSize()
public int getMinimumPoolSize()
public int getPoolSize()
public void interruptAll()
public void runWhenBlocked()
public void setKeepAliveTime(long msecs)
public void setMaximumPoolSize(int newMaximum)
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 void setMinimumPoolSize(int newMinimum)
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 void shutdownAfterProcessingCurrentlyQueuedTasks()
public void waitWhenBlocked()
|
Colt 1.0.2 | |||||||||
PREV CLASS NEXT CLASS | FRAMES NO FRAMES | |||||||||
SUMMARY: INNER | FIELD | CONSTR | METHOD | DETAIL: FIELD | CONSTR | METHOD |