org.jboss.netty.handler.execution

类 MemoryAwareThreadPoolExecutor

java.lang.Object

java.util.concurrent.AbstractExecutorService

java.util.concurrent.ThreadPoolExecutor

org.jboss.netty.handler.execution.MemoryAwareThreadPoolExecutor

所有已实现的接口:

java.util.concurrent.Executor, java.util.concurrent.ExecutorService

直接已知子类:

FairOrderedMemoryAwareThreadPoolExecutor, OrderedMemoryAwareThreadPoolExecutor

public class MemoryAwareThreadPoolExecutor
extends java.util.concurrent.ThreadPoolExecutor

A 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.

Using an alternative task size estimation strategy

Although the default implementation does its best to guess the size of an object of unknown type, it is always good idea to to use an alternative ObjectSizeEstimator implementation instead of the DefaultObjectSizeEstimator to avoid incorrect task size calculation, especially when:

• you are using MemoryAwareThreadPoolExecutor independently from ExecutionHandler,
• you are submitting a task whose type is not ChannelEventRunnable, or
• the message type of the MessageEvent in the ChannelEventRunnable is not ChannelBuffer.

Here is an example that demonstrates how to implement an ObjectSizeEstimator which understands a user-defined object:

 public class MyRunnable implements Runnable {

     private final byte[] data;

     public MyRunnable(byte[] data) {
         this.data = data;
     }

     public void run() {
         // Process 'data' ..
     }
 }

 public class MyObjectSizeEstimator extends DefaultObjectSizeEstimator {

     @Override
     public int estimateSize(Object o) {
         if (o instanceof MyRunnable) {
             return ((MyRunnable) o).data.length + 8;
         }
         return super.estimateSize(o);
     }
 }

 ThreadPoolExecutor pool = new MemoryAwareThreadPoolExecutor(
         16, 65536, 1048576, 30, TimeUnit.SECONDS,
         new MyObjectSizeEstimator(),
         Executors.defaultThreadFactory());

 pool.execute(new MyRunnable(data));
 

Event execution order

Please note that this executor does not maintain the order of the ChannelEvents 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.

嵌套类概要

从类继承的嵌套类/接口 java.util.concurrent.ThreadPoolExecutor

java.util.concurrent.ThreadPoolExecutor.AbortPolicy, java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy, java.util.concurrent.ThreadPoolExecutor.DiscardOldestPolicy, java.util.concurrent.ThreadPoolExecutor.DiscardPolicy

构造器概要

构造器

构造器和说明
MemoryAwareThreadPoolExecutor(int corePoolSize, long maxChannelMemorySize, long maxTotalMemorySize) Creates a new instance.
MemoryAwareThreadPoolExecutor(int corePoolSize, long maxChannelMemorySize, long maxTotalMemorySize, long keepAliveTime, java.util.concurrent.TimeUnit unit) Creates a new instance.
MemoryAwareThreadPoolExecutor(int corePoolSize, long maxChannelMemorySize, long maxTotalMemorySize, long keepAliveTime, java.util.concurrent.TimeUnit unit, ObjectSizeEstimator objectSizeEstimator, java.util.concurrent.ThreadFactory threadFactory) Creates a new instance.
MemoryAwareThreadPoolExecutor(int corePoolSize, long maxChannelMemorySize, long maxTotalMemorySize, long keepAliveTime, java.util.concurrent.TimeUnit unit, java.util.concurrent.ThreadFactory threadFactory) Creates a new instance.

方法概要

方法

限定符和类型	方法和说明
protected void	beforeExecute(java.lang.Thread t, java.lang.Runnable r)
protected void	decreaseCounter(java.lang.Runnable task)
protected void	doExecute(java.lang.Runnable task) Put the actual execution logic here.
protected void	doUnorderedExecute(java.lang.Runnable task) Executes the specified task without maintaining the event order.
void	execute(java.lang.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(java.lang.Runnable task)
boolean	remove(java.lang.Runnable task)
void	setMaxChannelMemorySize(long maxChannelMemorySize) Sets the maximum total size of the queued events per channel.
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(java.lang.Runnable task) Returns true if and only if the specified task should be counted to limit the global and per-channel memory consumption.
java.util.List<java.lang.Runnable>	shutdownNow() This will call shutdownNow(boolean) with the value of getNotifyChannelFuturesOnShutdown().
java.util.List<java.lang.Runnable>	shutdownNow(boolean notify) See ThreadPoolExecutor.shutdownNow() for how it handles the shutdown.
protected void	terminated()

从类继承的方法 java.util.concurrent.ThreadPoolExecutor

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

从类继承的方法 java.util.concurrent.AbstractExecutorService

invokeAll, invokeAll, invokeAny, invokeAny, newTaskFor, newTaskFor, submit, submit, submit

从类继承的方法 java.lang.Object

clone, equals, getClass, hashCode, notify, notifyAll, wait, wait, wait

构造器详细资料

MemoryAwareThreadPoolExecutor

public MemoryAwareThreadPoolExecutor(int corePoolSize,
                             long maxChannelMemorySize,
                             long maxTotalMemorySize)

Creates a new instance.

参数:

corePoolSize - the maximum number of active threads

maxChannelMemorySize - 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.

MemoryAwareThreadPoolExecutor

public MemoryAwareThreadPoolExecutor(int corePoolSize,
                             long maxChannelMemorySize,
                             long maxTotalMemorySize,
                             long keepAliveTime,
                             java.util.concurrent.TimeUnit unit)

Creates a new instance.

参数:

corePoolSize - the maximum number of active threads

maxChannelMemorySize - 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 down

unit - the TimeUnit of keepAliveTime

MemoryAwareThreadPoolExecutor

public MemoryAwareThreadPoolExecutor(int corePoolSize,
                             long maxChannelMemorySize,
                             long maxTotalMemorySize,
                             long keepAliveTime,
                             java.util.concurrent.TimeUnit unit,
                             java.util.concurrent.ThreadFactory threadFactory)

Creates a new instance.

参数:

corePoolSize - the maximum number of active threads

maxChannelMemorySize - 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 down

unit - the TimeUnit of keepAliveTime

threadFactory - the ThreadFactory of this pool

MemoryAwareThreadPoolExecutor

public MemoryAwareThreadPoolExecutor(int corePoolSize,
                             long maxChannelMemorySize,
                             long maxTotalMemorySize,
                             long keepAliveTime,
                             java.util.concurrent.TimeUnit unit,
                             ObjectSizeEstimator objectSizeEstimator,
                             java.util.concurrent.ThreadFactory threadFactory)

Creates a new instance.

参数:

corePoolSize - the maximum number of active threads

maxChannelMemorySize - 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 down

unit - the TimeUnit of keepAliveTime

threadFactory - the ThreadFactory of this pool

objectSizeEstimator - the ObjectSizeEstimator of this pool

方法详细资料

terminated

protected void terminated()

覆盖:

terminated 在类中 java.util.concurrent.ThreadPoolExecutor

shutdownNow

public java.util.List<java.lang.Runnable> shutdownNow()

This will call shutdownNow(boolean) with the value of getNotifyChannelFuturesOnShutdown().

指定者:

shutdownNow 在接口中 java.util.concurrent.ExecutorService

覆盖:

shutdownNow 在类中 java.util.concurrent.ThreadPoolExecutor

shutdownNow

public java.util.List<java.lang.Runnable> shutdownNow(boolean notify)

See 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.

getObjectSizeEstimator

public ObjectSizeEstimator getObjectSizeEstimator()

Returns the ObjectSizeEstimator of this pool.

setObjectSizeEstimator

public void setObjectSizeEstimator(ObjectSizeEstimator objectSizeEstimator)

Sets the ObjectSizeEstimator of this pool.

getMaxChannelMemorySize

public long getMaxChannelMemorySize()

Returns the maximum total size of the queued events per channel.

setMaxChannelMemorySize

public void setMaxChannelMemorySize(long maxChannelMemorySize)

Sets the maximum total size of the queued events per channel. Specify 0 to disable.

getMaxTotalMemorySize

public long getMaxTotalMemorySize()

Returns the maximum total size of the queued events for this pool.

setNotifyChannelFuturesOnShutdown

public void setNotifyChannelFuturesOnShutdown(boolean notifyOnShutdown)

If set to 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.

getNotifyChannelFuturesOnShutdown

public boolean getNotifyChannelFuturesOnShutdown()

Returns if the ChannelFuture's of the ChannelEventRunnable's should be notified about the shutdown of this MemoryAwareThreadPoolExecutor.

execute

public void execute(java.lang.Runnable command)

指定者:

execute 在接口中 java.util.concurrent.Executor

覆盖:

execute 在类中 java.util.concurrent.ThreadPoolExecutor

doExecute

protected void doExecute(java.lang.Runnable task)

Put the actual execution logic here. The default implementation simply calls doUnorderedExecute(Runnable).

doUnorderedExecute

protected final void doUnorderedExecute(java.lang.Runnable task)

Executes the specified task without maintaining the event order.

remove

public boolean remove(java.lang.Runnable task)

覆盖:

remove 在类中 java.util.concurrent.ThreadPoolExecutor

beforeExecute

protected void beforeExecute(java.lang.Thread t,
                 java.lang.Runnable r)

覆盖:

beforeExecute 在类中 java.util.concurrent.ThreadPoolExecutor

increaseCounter

protected void increaseCounter(java.lang.Runnable task)

decreaseCounter

protected void decreaseCounter(java.lang.Runnable task)

shouldCount

protected boolean shouldCount(java.lang.Runnable task)

Returns 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.