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 1) --- Channel A (Event 2) ---------------------------> 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.
|
boolean |
getNotifyChannelFuturesOnShutdown()
Returns if the
ChannelFuture 's of the ChannelEventRunnable 's should be
notified about the shutdown of this MemoryAwareThreadPoolExecutor . |
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 |
setNotifyChannelFuturesOnShutdown(boolean notifyOnShutdown)
If set to
false no queued ChannelEventRunnable 's ChannelFuture
will get notified once shutdownNow() is called. |
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. |
List<Runnable> |
shutdownNow()
This will call
shutdownNow(boolean) with the value of getNotifyChannelFuturesOnShutdown() . |
List<Runnable> |
shutdownNow(boolean notify)
See
ThreadPoolExecutor.shutdownNow() for how it handles the shutdown. |
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, 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 List<Runnable> shutdownNow()
shutdownNow(boolean)
with the value of getNotifyChannelFuturesOnShutdown()
.shutdownNow
in interface ExecutorService
shutdownNow
in class ThreadPoolExecutor
public List<Runnable> shutdownNow(boolean notify)
ThreadPoolExecutor.shutdownNow()
for how it handles the shutdown.
If true
is given to this method it also notifies all ChannelFuture
's
of the not executed ChannelEventRunnable
's.
Be aware that if you call this with false
you will need to handle the
notification of the ChannelFuture
's by your self. So only use this if you
really have a use-case for it.
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 setNotifyChannelFuturesOnShutdown(boolean notifyOnShutdown)
false
no queued ChannelEventRunnable
's ChannelFuture
will get notified once shutdownNow()
is called. If set to true
every
queued ChannelEventRunnable
will get marked as failed via ChannelFuture.setFailure(Throwable)
.
Please only set this to false
if you want to handle the notification by yourself
and know what you are doing. Default is true
.
public boolean getNotifyChannelFuturesOnShutdown()
ChannelFuture
's of the ChannelEventRunnable
's should be
notified about the shutdown of this MemoryAwareThreadPoolExecutor
.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-2014 The Netty Project. All Rights Reserved.