org.jboss.netty.handler.execution

类 OrderedMemoryAwareThreadPoolExecutor

java.lang.Object

java.util.concurrent.AbstractExecutorService

java.util.concurrent.ThreadPoolExecutor

org.jboss.netty.handler.execution.MemoryAwareThreadPoolExecutor

org.jboss.netty.handler.execution.OrderedMemoryAwareThreadPoolExecutor

所有已实现的接口:

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

直接已知子类:

OrderedDownstreamThreadPoolExecutor

public class OrderedMemoryAwareThreadPoolExecutor
extends MemoryAwareThreadPoolExecutor

A MemoryAwareThreadPoolExecutor which makes sure the events from the same Channel are executed sequentially.

NOTE: This thread pool inherits most characteristics of its super type, so please make sure to refer to MemoryAwareThreadPoolExecutor to understand how it works basically.

Event execution order

For example, let's say there are two executor threads that handle the events from the two channels:

           -------------------------------------> Timeline ------------------------------------>

 Thread X: --- Channel A (Event A1) --.   .-- Channel B (Event B2) --- Channel B (Event B3) --->
                                      \ /
                                       X
                                      / \
 Thread Y: --- Channel B (Event B1) --'   '-- Channel A (Event A2) --- Channel A (Event A3) --->
 

As you see, the events from different channels are independent from each other. That is, an event of Channel B will not be blocked by an event of Channel A and vice versa, unless the thread pool is exhausted.

Also, it is guaranteed that the invocation will be made sequentially for the events from the same channel. For example, the event A2 is never executed before the event A1 is finished. (Although not recommended, if you want the events from the same channel to be executed simultaneously, please use MemoryAwareThreadPoolExecutor instead.)

However, it is not guaranteed that the invocation will be made by the same thread for the same channel. The events from the same channel can be executed by different threads. For example, the Event A2 is executed by the thread Y while the event A1 was executed by the thread X.

Using a different key other than Channel to maintain event order

OrderedMemoryAwareThreadPoolExecutor uses a Channel as a key that is used for maintaining the event execution order, as explained in the previous section. Alternatively, you can extend it to change its behavior. For example, you can change the key to the remote IP of the peer:

 public class RemoteAddressBasedOMATPE extends OrderedMemoryAwareThreadPoolExecutor {

     ... Constructors ...

     @Override
     protected ConcurrentMap<Object, Executor> newChildExecutorMap() {
         // The default implementation returns a special ConcurrentMap that
         // uses identity comparison only (see IdentityHashMap).
         // Because SocketAddress does not work with identity comparison,
         // we need to employ more generic implementation.
         return new ConcurrentHashMap<Object, Executor>
     }

     protected Object getChildExecutorKey(ChannelEvent e) {
         // Use the IP of the remote peer as a key.
         return ((InetSocketAddress) e.getChannel().getRemoteAddress()).getAddress();
     }

     // Make public so that you can call from anywhere.
     public boolean removeChildExecutor(Object key) {
         super.removeChildExecutor(key);
     }
 }
 

Please be very careful of memory leak of the child executor map. You must call removeChildExecutor(Object) when the life cycle of the key ends (e.g. all connections from the same IP were closed.) Also, please keep in mind that the key can appear again after calling removeChildExecutor(Object) (e.g. a new connection could come in from the same old IP after removal.) If in doubt, prune the old unused or stall keys from the child executor map periodically:

 RemoteAddressBasedOMATPE executor = ...;

 on every 3 seconds:

   for (Iterator<Object> i = executor.getChildExecutorKeySet().iterator; i.hasNext();) {
       InetAddress ip = (InetAddress) i.next();
       if (there is no active connection from 'ip' now &&
           there has been no incoming connection from 'ip' for last 10 minutes) {
           i.remove();
       }
   }
 

If the expected maximum number of keys is small and deterministic, you could use a weak key map such as ConcurrentWeakHashMap or synchronized WeakHashMap instead of managing the life cycle of the keys by yourself.

嵌套类概要

嵌套类

限定符和类型	类和说明
protected class	OrderedMemoryAwareThreadPoolExecutor.ChildExecutor

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

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

字段概要

字段

限定符和类型	字段和说明
protected java.util.concurrent.ConcurrentMap<java.lang.Object,java.util.concurrent.Executor>	childExecutors

构造器概要

构造器

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

方法概要

方法

限定符和类型	方法和说明
protected void	doExecute(java.lang.Runnable task) Executes the specified task concurrently while maintaining the event order.
protected java.util.concurrent.Executor	getChildExecutor(ChannelEvent e)
protected java.lang.Object	getChildExecutorKey(ChannelEvent e)
protected java.util.Set<java.lang.Object>	getChildExecutorKeySet()
protected java.util.concurrent.ConcurrentMap<java.lang.Object,java.util.concurrent.Executor>	newChildExecutorMap()
protected boolean	removeChildExecutor(java.lang.Object key)
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.

从类继承的方法 org.jboss.netty.handler.execution.MemoryAwareThreadPoolExecutor

beforeExecute, decreaseCounter, doUnorderedExecute, execute, getMaxChannelMemorySize, getMaxTotalMemorySize, getNotifyChannelFuturesOnShutdown, getObjectSizeEstimator, increaseCounter, remove, setMaxChannelMemorySize, setNotifyChannelFuturesOnShutdown, setObjectSizeEstimator, shutdownNow, shutdownNow, 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

字段详细资料

childExecutors

protected final java.util.concurrent.ConcurrentMap<java.lang.Object,java.util.concurrent.Executor> childExecutors

构造器详细资料

OrderedMemoryAwareThreadPoolExecutor

public OrderedMemoryAwareThreadPoolExecutor(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.

OrderedMemoryAwareThreadPoolExecutor

public OrderedMemoryAwareThreadPoolExecutor(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

OrderedMemoryAwareThreadPoolExecutor

public OrderedMemoryAwareThreadPoolExecutor(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

OrderedMemoryAwareThreadPoolExecutor

public OrderedMemoryAwareThreadPoolExecutor(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

方法详细资料

newChildExecutorMap

protected java.util.concurrent.ConcurrentMap<java.lang.Object,java.util.concurrent.Executor> newChildExecutorMap()

getChildExecutorKey

protected java.lang.Object getChildExecutorKey(ChannelEvent e)

getChildExecutorKeySet

protected java.util.Set<java.lang.Object> getChildExecutorKeySet()

removeChildExecutor

protected boolean removeChildExecutor(java.lang.Object key)

doExecute

protected void doExecute(java.lang.Runnable task)

Executes the specified task concurrently while maintaining the event order.

覆盖:

doExecute 在类中 MemoryAwareThreadPoolExecutor

getChildExecutor

protected java.util.concurrent.Executor getChildExecutor(ChannelEvent e)

shouldCount

protected boolean shouldCount(java.lang.Runnable task)

从类复制的说明: MemoryAwareThreadPoolExecutor

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.

覆盖:

shouldCount 在类中 MemoryAwareThreadPoolExecutor