MINA线程模型

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

一、三种工作线程：

(一) Acceptor thread：

该线程的作用是接收客户端的连接，并将客户端的连接导入到IOProcessor线程模型中。Acceptor thread在调用了Acceptor.bind()方法后启动。IoAcceptor用于监听客户端的连接，每监听一个端口建立一个线程。每个Acceptor只能创建一个Acceptor thread，该线程模型不能配置（可以在Acceptor.bind时绑定多个端口，这样就能有多个Acceptor Thread，提高服务器对客户端连接的效率），它由Mina自身提供。

(二) Connector thread：

该线程模型是客户端的连接线程模型，它的作用和Acceptor thread类似，它将客户端与服务器的连接导入到IOProcessor线程模型中。IoConnector用于与服务端建立连接，每连接一个服务端就建立一个线程。同样地，该线程模型也是由Mina的客户端自动创建，该线程模型也不能进行配置。

(三) IOProcessor thread：

该线程模型的主要作用就是接收和发送数据，所有的IO操作在服务器与客户端的连接建立后，所有的数据的接收和发送都是有该线程模型来负责的，直到客户端与服务器的连接关闭，该线程模型才停止工作。该线程模型可以由程序员根据需要进行配置。该线程模型默认的线程的数量为cpu的核数+1。若你的cpu为双核的，则你的I/O processor线程的最大数量为3，同理若你的若你的cpu为四核的，那么你的I/O processor线程的最大数量为5。

由上面的内容我们可以知道在Mina中可以配置的线程数量只有IOProcessor，对于每个IoService在创建其实例的时候可以配置该IoService的IOProcessor的线程数量。在

SokcetConnector和SocketAccpetor中IOPProcessor的数量是由CPU的核数+1来决定的。

二、线程模型：

单一的Processor线程内部来看，IO请求的处理流程是单线程顺序处理的。当Process线程select了一批就绪的IO请求后，会在线程内部逐一对这些IO请求进行处理。处理的流程包括IoFilter和IoHandler里的逻辑。当前面的IO请求处理完毕后，才会取下一个IO请求进行处理。也就是说，如果IoFilter或IoHandler中有比较耗时的操作的话（如：读取数据库等），Processor线程将会被阻塞住，后续的请求将得不到处理。这样的情况在高并发的服务器下显然是不能容忍的。于是，Mina通过在处理流程中引入线程池来解决这个问题。

IoFilterChain是Mina的扩展点。Mina里是通过IoFilter的形式来为处理流程添加线程池的。Mina的线程模型主要有一下这几种形式：

1、单线程模型，也是Mina默认线程模型。也就是Processor包办了从底层IO到上层的IoHandler逻辑的所有执行工作。这种模型比较适合于处理逻辑简单，能快速返回的情况。

2、在IoFilterChain中加入了Thread Pool Filter。此时的处理流程变为Processor线程读取完数据后，执行IoFilterChain的逻辑。当执行到Thread Pool Filter的时候，该Filter会将后续的处理流程封装到一个Runnable对象中，并交由Filter自身的线程池来执行，而Processor线程则能立即返回来处理下一个IO请求。这样如果后面的IoFilter或IoHandler中有阻塞操作，只会引起Filter线程池里的线程阻塞，而不会阻塞住Processor线程，从而提高了服务器的处理能力。Mina提供了Thread Pool Filter的一个实现：ExecutorFilter。

3、没有限制说chain中只能添加一个ExecutorFilter，开发者也可以在chain中加入多个ExecutorFilter来构成第三种情况，但一般情况下可能没有这个必要。

三、ExecutorFilter：

添加ExecutorFilter到IOFilterChain中的代码如下：

acceptor.getFilterChain().addLast("threadPool", new ExecutorFilter(Executors.newCachedThreadPool()));

ExecutorFilter创建时的主要参数有：

1、指定线程池的属性信息，譬如：核心大小、最大大小、等待队列的性质等。特别要关注的是ExecutorFilter 内部默认使用的是OrderedThreadPoolExecutor 作为线程池的实现，从名字上可以看出是保证各个事件在多线程执行中的顺序；

2、哪些事件方法被关注，也就哪些事件方法用这个线程池执行。线程池可以异步执行的事件类型是位于IoEventType 中的九个枚举值中除了SESSION_CREATED 之外的其余八个，这说明Session 建立的事件只能与IoProcessor 在同一个线程上执行。

默认情况下，没有配置关注的事件类型，有如下六个事件方法会被自动使用线程池异步执行：

IoEventType.EXCEPTION_CAUGHT,
IoEventType.MESSAGE_RECEIVED,
IoEventType.MESSAGE_SENT,
IoEventType.SESSION_CLOSED,
IoEventType.SESSION_IDLE,
IoEventType.SESSION_OPENED

3、绝对不能开启线程让其执行sessionCreated()方法。如果你真的打算使用这个ExecutorFilter，那么最好想清楚它该放在过滤器链的哪个位置，针对哪些事件做异步处理机制。一般ExecutorFilter 都是要放在ProtocolCodecFilter 过滤器的后面，也就是不要让编解码运行在独立的线程上，而是要运行在IoProcessor 所在的线程，因为编解码处理的数据都是由IoProcessor 读取和发送的，没必要开启新的线程，否则性能反而会下降。一般使用ExecutorFilter 的典型场景是将业务逻辑（譬如：耗时的数据库操作）放在单独的线程中运行，也就是说与IO 处理无关的操作可以考虑使用ExecutorFilter 来异步执行。

4、请求的处理顺序:

在处理流程中加入线程池，可以较好的提高服务器的吞吐量，但也带来了新的问题：请求的处理顺序问题。在单线程的模型下，可以保证IO请求是挨个顺序地处理的。加入线程池之后，同一个IoSession的多个IO请求可能被ExecutorFilter并行的处理，这对于一些对请求处理顺序有要求的程序来说是不希望看到的。比如：数据库服务器处理同一个会话里的prepare，execute，commit请求希望是能按顺序逐一执行的。

Mina里默认的实现是有保证同一个IoSession中IO请求的顺序的。具体的实现是，ExecutorFilter默认采用了Mina提供的OrderedThreadPoolExecutor作为内置线程池。后者并不会立即执行加入进来的Runnable对象，而是会先从Runnable对象里获取关联的IoSession(这里有个down cast成IoEvent的操作)，并将Runnable对象加入到session的任务列表中。OrderedThreadPoolExecutor会按session里任务列表的顺序来处理请求，从而保证了请求的执行顺序。

对于没有顺序要请求的情况，可以为ExecutorFilter指定一个Executor来替换掉默认的OrderedThreadPoolExecutor，让同一个session的多个请求能被并行地处理，来进一步提高吞吐量。

5、ExecutorFilter 的工作机制很简单，就是在调用下一个过滤器的事件方法时，封装成IoFilterEvent对象并把其交给Executor 的execute(Runnable runnable)方法来执行：

………………
@Override
public final void messageReceived(NextFilter nextFilter, IoSession session, Object message) {
if (eventTypes.contains(IoEventType.MESSAGE_RECEIVED)) {
IoFilterEvent event = new IoFilterEvent(nextFilter, IoEventType.MESSAGE_RECEIVED, session, message);
fireEvent(event);
} else {
nextFilter.messageReceived(session, message);
}
}
@Override
public final void messageSent(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) {
if (eventTypes.contains(IoEventType.MESSAGE_SENT)) {
IoFilterEvent event = new IoFilterEvent(nextFilter, IoEventType.MESSAGE_SENT, session, writeRequest);
fireEvent(event);
} else {
nextFilter.messageSent(session, writeRequest);
}
}
……………… protected void fireEvent(IoFilterEvent event) {
executor.execute(event);
}