【MINA】OrderedThreadPoolExecutor和UnorderedThreadPoolExecutor的事件监听线程池的选择

mina中有两个线程池概念 1.处理监听建立连接的线程池 2.处理读写事件的线程池

本文中主要探讨读写事件的线程池的选择

这两种都经过实际项目的使用和检测，说说优缺点

早期的项目是用UnorderedThreadPoolExecutor【无序线程池】

特点：线程池管理[b]一个无界阻塞队列，线程在分配事件，并发处理maximumPoolSize个事件，[/b]

后面来的事件在没有空闲线程的时候要进入这个无界阻塞队列等待执行

可以看出，他的特点是session之间是没有影响的，从事件的执行效率上要更高一些，但是同session的事件执行顺序很难保证

UnorderedThreadPoolExecutor源码

public UnorderedThreadPoolExecutor(
int corePoolSize, int maximumPoolSize,
long keepAliveTime, TimeUnit unit,
ThreadFactory threadFactory, IoEventQueueHandler queueHandler) {

super(0, 1, keepAliveTime, unit, new LinkedBlockingQueue<Runnable>(), threadFactory, new AbortPolicy());
if (corePoolSize < 0) {
throw new IllegalArgumentException("corePoolSize: " + corePoolSize);
}

if (maximumPoolSize == 0 || maximumPoolSize < corePoolSize) {
throw new IllegalArgumentException("maximumPoolSize: " + maximumPoolSize);
}

if (queueHandler == null) {
queueHandler = IoEventQueueHandler.NOOP;
}

this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.queueHandler = queueHandler;
}

@Override
public void execute(Runnable task) {
if (shutdown) {
rejectTask(task);
}

checkTaskType(task);

IoEvent e = (IoEvent) task;
boolean offeredEvent = queueHandler.accept(this, e);
if (offeredEvent) {
getQueue().offer(e);
}

addWorkerIfNecessary();

if (offeredEvent) {
queueHandler.offered(this, e);
}
}

　　

后来的新项目使用了OrderedThreadPoolExecutor【有序的线程池】

特点：每个session有自己的事件队列，线程池在分配session，相当于每个session有唯一的线程在处理session中的任务队列，如果没有空闲线程可调度，

那么其他session就只能进入waitingSessions这个阻塞队列等待有空闲线程接管这个session的任务队列

可以看出，session之间是有影响的，但是保证了同session的事件执行是有序的

MyOrderedThreadPoolExecutor代码完全抄OrderedThreadPoolExecutor目地是想加入一个参数

maxQueueSize控制每个session任务队列的最大长度，超出就丢弃后进的任务

public MyOrderedThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
long keepAliveTime, TimeUnit unit, ThreadFactory threadFactory,
IoEventQueueHandler eventQueueHandler, int maxQueueSize) {
// 从这里可以看到,其实现类似于{@link Executors.newCachedThreadPool},不过其可以设置corePoolSize和maximumPoolSize.

// 不过其execute方法是自实现的，否则如果用父类的则会出现问题，即在任务繁忙的时候会出现任务被拒绝

// 因为其把任务放到了session的任务队列中.即没有由线程池本身来保存

// 另外可以看到初始化的corePoolSize和maximumPoolSize分别传了0和1.这是为了更好的处理异常,因为super不能try/catch

super(DEFAULT_INITIAL_THREAD_POOL_SIZE, 1, keepAliveTime, unit,
new SynchronousQueue<Runnable>(), threadFactory,
new AbortPolicy());

if (corePoolSize < DEFAULT_INITIAL_THREAD_POOL_SIZE) {
throw new IllegalArgumentException("corePoolSize: " + corePoolSize);
}

if ((maximumPoolSize == 0) || (maximumPoolSize < corePoolSize)) {
throw new IllegalArgumentException("maximumPoolSize: "
+ maximumPoolSize);
}

// Now, we can setup the pool sizes
super.setCorePoolSize(corePoolSize);
super.setMaximumPoolSize(maximumPoolSize);

// The queueHandler might be null.
if (eventQueueHandler == null) {
this.eventQueueHandler = IoEventQueueHandler.NOOP;
} else {
this.eventQueueHandler = eventQueueHandler;
}

this.maxQueueSize = maxQueueSize;
}

@Override
public void execute(Runnable task) {
if (shutdown) {
rejectTask(task);
}

// Check that it's a IoEvent task
checkTaskType(task);

IoEvent event = (IoEvent) task;

// Get the associated session
IoSession session = event.getSession();

// Get the session's queue of events
SessionTasksQueue sessionTasksQueue = getSessionTasksQueue(session);
Queue<Runnable> tasksQueue = sessionTasksQueue.tasksQueue;

　　　　　　　　　//超过任务队列上限就丢弃任务，目前定的值时20
if (tasksQueue.size() > this.maxQueueSize) {
LOGGER.error(com.youxigu.dynasty2.i18n.MarkupMessages.getString("MyOrderedThreadPoolExecutor_0"), session.getRemoteAddress(),
this.maxQueueSize);
rejectTask(task);
return;
}
boolean offerSession;

// propose the new event to the event queue handler. If we
// use a throttle queue handler, the message may be rejected
// if the maximum size has been reached.
boolean offerEvent = eventQueueHandler.accept(this, event);

if (offerEvent) {
// Ok, the message has been accepted
synchronized (tasksQueue) {
// Inject the event into the executor taskQueue
tasksQueue.offer(event);

if (sessionTasksQueue.processingCompleted) {
sessionTasksQueue.processingCompleted = false;
offerSession = true;
} else {
offerSession = false;
}

if (LOGGER.isDebugEnabled()) {
print(tasksQueue, event);
}
}
} else {
offerSession = false;
}

if (offerSession) {
// As the tasksQueue was empty, the task has been executed
// immediately, so we can move the session to the queue
// of sessions waiting for completion.
waitingSessions.offer(session);
}

addWorkerIfNecessary();

if (offerEvent) {
eventQueueHandler.offered(this, event);
}
}

　　

1.为什么要选用后者那

在业务层通常是有locker控制并发，但是还是可能产生并发事件，导致数据错乱的问题，

1>同session的事件执行顺序，不能保证，

2>程序员写业务时对locker的使用不规范，多对象修改时考虑不全，但是单线程有序执行，就可以解决这些问题，但对不同session导致数据交叉是没作用的，那个还得依赖locker

2.考虑两个问题

1>session队列的长度要可控，所以要改造一下，见上面MyOrderedThreadPoolExecutor 红色代码部分

2>线程初始数量的上限的定义选值

3>ExecutorFilter提供了业务逻辑的执行线程，可以将应用层业务逻辑通过配置该filter在配置的线程池内执行

/**
* 消息执行器线程池的最小线程数
*/
private int corePoolSize = Runtime.getRuntime().availableProcessors() * 4;

/**
* 消息执行器线程池的最大线程数，读写分开是为了防止大量的读请求阻塞回写
*/
private int maxPoolSize = Runtime.getRuntime().availableProcessors() * 24 + 1;

/**
* 消息执行器线程池超过corePoolSize的Thread存活时间;秒
*/
private long keepAliveTime = 300;

　　　　　　/**
　　　　　　* useOrderedPool=true的时候有效,指定一个session的请求队列的最大长度，超过这个长度的请求丢弃.
　　　　　　*/
　　　　　private int orderedPoolMaxQueueSize = 20;

略。。。。。

ThreadPoolExecutor executor = new MyOrderedThreadPoolExecutor(corePoolSize,
maxPoolSize, keepAliveTime, TimeUnit.SECONDS,
new NamingThreadFactory("WolfServerThread"),
ioEventQueueHandler, orderedPoolMaxQueueSize);

　　　　　　　　　// 这里是预先启动corePoolSize个处理线程
executor.prestartAllCoreThreads();

　　　　　　　　　//这里指定的监听事件就会由这里指定的线程池来处，而不在是processr线程来处理
chain.addLast("exec", new ExecutorFilter(executor,
IoEventType.EXCEPTION_CAUGHT, IoEventType.MESSAGE_RECEIVED,
IoEventType.SESSION_CLOSED, IoEventType.SESSION_IDLE,
IoEventType.SESSION_OPENED));

一篇好文章讲ExecutorFilter http://www.blogjava.net/landon/archive/2014/02/03/409521.html