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

直接已知子类：

OrderedMemoryAwareThreadPoolExecutor

public class MemoryAwareThreadPoolExecutor
extends java.util.concurrent.ThreadPoolExecutor

一个当队列里有太多的任务时阻塞任务提交的ThreadPoolExecutor.每Channel和每 Executor限制会被应用.

当一个任务(如. Runnable)被提交,MemoryAwareThreadPoolExecutor调用 ObjectSizeEstimator.estimateSize(Object)获取任务的字节估算大小去计算被未执行任务占用的内存总数.

如果未执行任务的总额大小超过了每Channel或每Executor的阀值,任何 execute(Runnable)的调用都会阻塞直到队列里的任务被执行以便总数小于阀值为止.

使用其他任务大小估算策略

尽管默认实现可以尽量猜测一个未知类型对象的大小,但可以使用另一个ObjectSizeEstimator实现代替 DefaultObjectSizeEstimator来避免不正确任务大小的计算,特别当:

• 你使用独立于ExecutionHandler的MemoryAwareThreadPoolExecutor (注:一般两者都是成对出现)
• 你提交一个类型不是ChannelEventRunnable的任务或
• 在ChannelEventRunnable里的MessageEvent的信息类型不是 ChannelBuffer.

这里的例子演示了如何实现一个能理解用户定义的对象的ObjectSizeEstimator:

 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));
 

事件执行顺序

请注意执行器并不维护同一个Channel的ChannelEvent的顺序.例如,你甚至可以在一个 "messageReceived"事件之前接收"channelClosed"事件,如下图所述. 例如,事件会如下图描述被执行:

           --------------------------------> 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) --->
 

要维护事件顺序,你必须使用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)
创建一个实例.

MemoryAwareThreadPoolExecutor(int corePoolSize, long maxChannelMemorySize, long maxTotalMemorySize, long keepAliveTime, java.util.concurrent.TimeUnit unit)
创建一个实例 .

MemoryAwareThreadPoolExecutor(int corePoolSize, long maxChannelMemorySize, long maxTotalMemorySize, long keepAliveTime, java.util.concurrent.TimeUnit unit, ObjectSizeEstimator objectSizeEstimator, java.util.concurrent.ThreadFactory threadFactory)
创建一个实例 .

MemoryAwareThreadPoolExecutor(int corePoolSize, long maxChannelMemorySize, long maxTotalMemorySize, long keepAliveTime, java.util.concurrent.TimeUnit unit, java.util.concurrent.ThreadFactory threadFactory)
创建一个实例 .

方法摘要

void	execute(java.lang.Runnable command)
long	getMaxChannelMemorySize() 返回每个通道队列事件的最大总大小.
long	getMaxTotalMemorySize() 返回该池的队列事件最大总大小.
ObjectSizeEstimator	getObjectSizeEstimator() 返回该池的ObjectSizeEstimator.
boolean	remove(java.lang.Runnable task)
void	setMaxChannelMemorySize(long maxChannelMemorySize) 设置每个通道队列事件的最大总大小.0为禁止.
void	setMaxTotalMemorySize(long maxTotalMemorySize) 设置该池队列事件的最大从大小,设置0为禁止.
void	setObjectSizeEstimator(ObjectSizeEstimator objectSizeEstimator) 设置该池的ObjectSizeEstimator.

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

allowCoreThreadTimeOut, allowsCoreThreadTimeOut, awaitTermination, getActiveCount, getCompletedTaskCount, getCorePoolSize, getKeepAliveTime, getLargestPoolSize, getMaximumPoolSize, getPoolSize, getQueue, getRejectedExecutionHandler, getTaskCount, getThreadFactory, isShutdown, isTerminated, isTerminating, prestartAllCoreThreads, prestartCoreThread, purge, setCorePoolSize, setKeepAliveTime, setMaximumPoolSize, setRejectedExecutionHandler, setThreadFactory, shutdown, shutdownNow

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

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

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

equals, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

构造方法详细信息

MemoryAwareThreadPoolExecutor

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

创建一个实例.

参数：

corePoolSize - 活动线程的最大数

maxChannelMemorySize - 每个通道队列事件最大总大小.0则禁止.

maxTotalMemorySize - 该池的队列事件最大总大小.0则禁止.

MemoryAwareThreadPoolExecutor

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

创建一个实例 .

参数：

corePoolSize - 活动线程的最大数

maxChannelMemorySize - 每个通道队列事件最大总大小.0则禁止.

maxTotalMemorySize - 该池的队列事件最大总大小.0则禁止.

keepAliveTime - 一个非活动线程关闭自己时间的时间总数

unit - keepAliveTime的时间单位

MemoryAwareThreadPoolExecutor

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

创建一个实例 .

参数：

corePoolSize - 活动线程的最大数

maxChannelMemorySize - 每个通道队列事件最大总大小.0则禁止.

maxTotalMemorySize - 该池的队列事件最大总大小.0则禁止.

keepAliveTime - 一个非活动线程关闭自己时间的时间总数

unit - keepAliveTime的时间单位

threadFactory - 该池的ThreadFactory

MemoryAwareThreadPoolExecutor

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

创建一个实例 .

参数：

corePoolSize - 活动线程的最大数

maxChannelMemorySize - 每个通道队列事件最大总大小.0则禁止.

maxTotalMemorySize - 该池的队列事件最大总大小.0则禁止.

keepAliveTime - 一个非活动线程关闭自己时间的时间总数

unit - keepAliveTime的时间单位

threadFactory - 该池的ThreadFactory

objectSizeEstimator - 该池的ObjectSizeEstimator

方法详细信息

getObjectSizeEstimator

public ObjectSizeEstimator getObjectSizeEstimator()

返回该池的ObjectSizeEstimator.

setObjectSizeEstimator

public void setObjectSizeEstimator(ObjectSizeEstimator objectSizeEstimator)

设置该池的ObjectSizeEstimator.

getMaxChannelMemorySize

public long getMaxChannelMemorySize()

返回每个通道队列事件的最大总大小.

setMaxChannelMemorySize

public void setMaxChannelMemorySize(long maxChannelMemorySize)

设置每个通道队列事件的最大总大小.0为禁止.

getMaxTotalMemorySize

public long getMaxTotalMemorySize()

返回该池的队列事件最大总大小.

setMaxTotalMemorySize

public void setMaxTotalMemorySize(long maxTotalMemorySize)

设置该池队列事件的最大从大小,设置0为禁止.

execute

public void execute(java.lang.Runnable command)

指定者：

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

覆盖：

类 java.util.concurrent.ThreadPoolExecutor 中的 execute

remove

public boolean remove(java.lang.Runnable task)

覆盖：

类 java.util.concurrent.ThreadPoolExecutor 中的 remove