public class MemoryAwareThreadPoolExecutor extends ThreadPoolExecutor
ThreadPoolExecutor
which blocks the task submission when there's
too many tasks in the queue. Both per-Channel
and per-Executor
limitation can be applied.
When a task (i.e. Runnable
) is submitted,
MemoryAwareThreadPoolExecutor
calls ObjectSizeEstimator.estimateSize(Object)
to get the estimated size of the task in bytes to calculate the amount of
memory occupied by the unprocessed tasks.
If the total size of the unprocessed tasks exceeds either per-Channel
or per-Executor
threshold, any further execute(Runnable)
call will block until the tasks in the queue are processed so that the total
size goes under the threshold.
ObjectSizeEstimator
implementation instead of the
DefaultObjectSizeEstimator
to avoid incorrect task size calculation,
especially when:
MemoryAwareThreadPoolExecutor
independently from
ExecutionHandler
,ChannelEventRunnable
, orMessageEvent
in the ChannelEventRunnable
is not ChannelBuffer
.ObjectSizeEstimator
which understands a user-defined object:
public class MyRunnable implementsRunnable
{ private final byte[] data; public MyRunnable(byte[] data) { this.data = data; } public void run() { // Process 'data' .. } } public class MyObjectSizeEstimator extendsDefaultObjectSizeEstimator
{ @Override public int estimateSize(Object o) { if (o instanceof MyRunnable) { return ((MyRunnable) o).data.length + 8; } return super.estimateSize(o); } }ThreadPoolExecutor
pool = newMemoryAwareThreadPoolExecutor
( 16, 65536, 1048576, 30,TimeUnit
.SECONDS, new MyObjectSizeEstimator(),Executors
.defaultThreadFactory()); pool.execute(new MyRunnable(data));
ChannelEvent
s for the same Channel
. For example,
you can even receive a "channelClosed"
event before a
"messageReceived"
event, as depicted by the following diagram.
For example, the events can be processed as depicted below:
--------------------------------> Timeline --------------------------------> Thread X: --- Channel A (Event 2) --- Channel A (Event 1) ---------------------------> Thread Y: --- Channel A (Event 3) --- Channel B (Event 2) --- Channel B (Event 3) ---> Thread Z: --- Channel B (Event 1) --- Channel B (Event 4) --- Channel A (Event 4) --->To maintain the event order, you must use
OrderedMemoryAwareThreadPoolExecutor
.ThreadPoolExecutor.AbortPolicy, ThreadPoolExecutor.CallerRunsPolicy, ThreadPoolExecutor.DiscardOldestPolicy, ThreadPoolExecutor.DiscardPolicy
Constructor and Description |
---|
MemoryAwareThreadPoolExecutor(int corePoolSize,
long maxChannelMemorySize,
long maxTotalMemorySize)
Creates a new instance.
|
MemoryAwareThreadPoolExecutor(int corePoolSize,
long maxChannelMemorySize,
long maxTotalMemorySize,
long keepAliveTime,
TimeUnit unit)
Creates a new instance.
|
MemoryAwareThreadPoolExecutor(int corePoolSize,
long maxChannelMemorySize,
long maxTotalMemorySize,
long keepAliveTime,
TimeUnit unit,
ObjectSizeEstimator objectSizeEstimator,
ThreadFactory threadFactory)
Creates a new instance.
|
MemoryAwareThreadPoolExecutor(int corePoolSize,
long maxChannelMemorySize,
long maxTotalMemorySize,
long keepAliveTime,
TimeUnit unit,
ThreadFactory threadFactory)
Creates a new instance.
|
Modifier and Type | Method and Description |
---|---|
protected void |
beforeExecute(Thread t,
Runnable r) |
protected void |
decreaseCounter(Runnable task) |
protected void |
doExecute(Runnable task)
Put the actual execution logic here.
|
protected void |
doUnorderedExecute(Runnable task)
Executes the specified task without maintaining the event order.
|
void |
execute(Runnable command) |
long |
getMaxChannelMemorySize()
Returns the maximum total size of the queued events per channel.
|
long |
getMaxTotalMemorySize()
Returns the maximum total size of the queued events for this pool.
|
ObjectSizeEstimator |
getObjectSizeEstimator()
Returns the
ObjectSizeEstimator of this pool. |
protected void |
increaseCounter(Runnable task) |
boolean |
remove(Runnable task) |
void |
setMaxChannelMemorySize(long maxChannelMemorySize)
Sets the maximum total size of the queued events per channel.
|
void |
setMaxTotalMemorySize(long maxTotalMemorySize)
Deprecated.
maxTotalMemorySize is not modifiable anymore.
|
void |
setObjectSizeEstimator(ObjectSizeEstimator objectSizeEstimator)
Sets the
ObjectSizeEstimator of this pool. |
protected boolean |
shouldCount(Runnable task)
Returns
true if and only if the specified task should
be counted to limit the global and per-channel memory consumption. |
protected void |
terminated() |
afterExecute, allowCoreThreadTimeOut, allowsCoreThreadTimeOut, awaitTermination, finalize, getActiveCount, getCompletedTaskCount, getCorePoolSize, getKeepAliveTime, getLargestPoolSize, getMaximumPoolSize, getPoolSize, getQueue, getRejectedExecutionHandler, getTaskCount, getThreadFactory, isShutdown, isTerminated, isTerminating, prestartAllCoreThreads, prestartCoreThread, purge, setCorePoolSize, setKeepAliveTime, setMaximumPoolSize, setRejectedExecutionHandler, setThreadFactory, shutdown, shutdownNow, toString
invokeAll, invokeAll, invokeAny, invokeAny, newTaskFor, newTaskFor, submit, submit, submit
public MemoryAwareThreadPoolExecutor(int corePoolSize, long maxChannelMemorySize, long maxTotalMemorySize)
corePoolSize
- the maximum number of active threadsmaxChannelMemorySize
- the maximum total size of the queued events per channel.
Specify 0
to disable.maxTotalMemorySize
- the maximum total size of the queued events for this pool
Specify 0
to disable.public MemoryAwareThreadPoolExecutor(int corePoolSize, long maxChannelMemorySize, long maxTotalMemorySize, long keepAliveTime, TimeUnit unit)
corePoolSize
- the maximum number of active threadsmaxChannelMemorySize
- the maximum total size of the queued events per channel.
Specify 0
to disable.maxTotalMemorySize
- the maximum total size of the queued events for this pool
Specify 0
to disable.keepAliveTime
- the amount of time for an inactive thread to shut itself downunit
- the TimeUnit
of keepAliveTime
public MemoryAwareThreadPoolExecutor(int corePoolSize, long maxChannelMemorySize, long maxTotalMemorySize, long keepAliveTime, TimeUnit unit, ThreadFactory threadFactory)
corePoolSize
- the maximum number of active threadsmaxChannelMemorySize
- the maximum total size of the queued events per channel.
Specify 0
to disable.maxTotalMemorySize
- the maximum total size of the queued events for this pool
Specify 0
to disable.keepAliveTime
- the amount of time for an inactive thread to shut itself downunit
- the TimeUnit
of keepAliveTime
threadFactory
- the ThreadFactory
of this poolpublic MemoryAwareThreadPoolExecutor(int corePoolSize, long maxChannelMemorySize, long maxTotalMemorySize, long keepAliveTime, TimeUnit unit, ObjectSizeEstimator objectSizeEstimator, ThreadFactory threadFactory)
corePoolSize
- the maximum number of active threadsmaxChannelMemorySize
- the maximum total size of the queued events per channel.
Specify 0
to disable.maxTotalMemorySize
- the maximum total size of the queued events for this pool
Specify 0
to disable.keepAliveTime
- the amount of time for an inactive thread to shut itself downunit
- the TimeUnit
of keepAliveTime
threadFactory
- the ThreadFactory
of this poolobjectSizeEstimator
- the ObjectSizeEstimator
of this poolprotected void terminated()
terminated
in class ThreadPoolExecutor
public ObjectSizeEstimator getObjectSizeEstimator()
ObjectSizeEstimator
of this pool.public void setObjectSizeEstimator(ObjectSizeEstimator objectSizeEstimator)
ObjectSizeEstimator
of this pool.public long getMaxChannelMemorySize()
public void setMaxChannelMemorySize(long maxChannelMemorySize)
0
to disable.public long getMaxTotalMemorySize()
@Deprecated public void setMaxTotalMemorySize(long maxTotalMemorySize)
public void execute(Runnable command)
execute
in interface Executor
execute
in class ThreadPoolExecutor
protected void doExecute(Runnable task)
doUnorderedExecute(Runnable)
.protected final void doUnorderedExecute(Runnable task)
public boolean remove(Runnable task)
remove
in class ThreadPoolExecutor
protected void beforeExecute(Thread t, Runnable r)
beforeExecute
in class ThreadPoolExecutor
protected void increaseCounter(Runnable task)
protected void decreaseCounter(Runnable task)
protected boolean shouldCount(Runnable task)
true
if and only if the specified task
should
be counted to limit the global and per-channel memory consumption.
To override this method, you must call super.shouldCount()
to
make sure important tasks are not counted.Copyright © 2008-2012 JBoss, a division of Red Hat, Inc.. All Rights Reserved.