MINA2 核心源代码分析笔记 - 2. IoAcceptor与IoConnector类族 (IoService)

 

2008年7月的一个周末，突发热忱，对刚出了M3版本的MINA2起了兴趣。下载了原码以及一部分资料进行研学。当时，做了一部分笔记。但是由于工作等各种原因一年多没有继续。正直2010年新年新春之际，于家中整理当时的笔记。准备后续慢慢整理出来。

当时的nio网络框架并不多，除了MINA2以外还有国内著名的Cindy。两个框架都作了研学，但是Cindy由于代码注释少、以及资料缺乏，作者也消失良久等原因，所以并没有作为重点学习目标。

经过3天对MINA2代码研读，发现的核心部分从来没有一篇doc能够说明清除（至少当时没有看到），而大部分都停留在使用IoHanlder以及IoFilter等外部扩展上。而我将以NIO的网络部分为切入点来介绍其真实的核心部分。

整个笔记会分为几部分进行说明：

1. 整体结构及代码分析指引

2. IoAcceptor与IoConnector类族 (IoService)

 

1 IoAcceptor类族

IoAcceptor类族表示接受socket连接并驱动IoProcessor的实体。概念上IoAcceptor与IoConnector形成IoService的功能，即表示NIO的所有服务。
其类族中，NET相关的类封装了Selector以及ServerSocketChannel，完成注册ServerSocketChannel到Selector并接受新到的SocketChannel。
NET相关主要包括如下接口与类：
AbstractIoAcceptor

AbstractPollingIoAcceptor

NioSocketAcceptor

其中AbstractPollingIoAcceptor类实现了核心的流转功能。NioSocketAcceptor实现了具体的网络部分。
AbstractPollingIoAcceptor主要封装了一个Worker 线程程序以及registerQueue、cancalQueue。Worker程序完成Java nio底层的selector的select()行为操作。而具体的selector封装在NioSocketAcceptor中。
AbstractPollingIoAcceptor的两个Queue如下：

//准备注册queue
private final Queue<AcceptorOperationFuture> registerQueue =
new ConcurrentLinkedQueue<AcceptorOperationFuture>();

//取消queue
private final Queue<AcceptorOperationFuture> cancelQueue =
new ConcurrentLinkedQueue<AcceptorOperationFuture>();

 
 
registerQueue是存放ServerSocketChannel的queue。Accptor进行侦听时生成ServerSocketChannel放入其中。Worker循环程序不断将其取出并注册到selector中。
cancelQueue是存放ServerSocketChannel的queue。Worker循环程序不断将其取出并关闭释放该ServerSocketChannel。
AbstractPollingIoProcessor的Worker核心代码如下：
 

while (selectable) {
try {

// gets the number of keys that are ready to go
boolean selected = select();
//1 注册op_accept
// this actually sets the selector to OP_ACCEPT,
// and binds to the port in which this class will
// listen on
nHandles += registerHandles();
//2 处理
if (selected) {
processHandles(selectedHandles());
}
//3 注销
// check to see if any cancellation request has been made.
nHandles -= unregisterHandles();
//
if (nHandles == 0) {
……
}
} catch (Throwable e) {
……
}
}

 

这个是一个典型的java
nio selector 循环。
select()用于等待OP_ACCEPT事件。

registerHandles()调用主要完成从registerQueue取出SSC并注册到selector。

processHandles(selectedHandles())主要完成生成新的IoSession对象并将其推送到IoProcessor。

unregisterHandles()主要是取出cancelQueue中的SSC，然后关闭。

 

 

 

2  IoConnector类族

与IoAccptor类族提供服务端（被动）功能类似，客户端（主动）部分由IoConnector类族提供功能。其实现方式也与IoAcceptor如出一辙。

其类族中，NET相关的IoConnector类族表示主动发起socket连接并驱动一个IoProcessor对象的实体。概念上IoAcceptor与IoConnector形成IoService的功能，即表示NIO的所有服务。
该类族主要包括以下接口和类：
AbstractIoConnector

AbstractPollingIoConnector

NioSocketConnector

其中AbstractPollingIoConnector是流转功能的核心类。其中分装了一个selector用以注册“CONNECT动作”。当完成连接时将IoSession推送到关联的IoProcessor 对象中进行下一步处理。
AbstractPollingIoConnector内部有如下Queue：

// 连接请求 连接队列
private final Queue<ConnectionRequest> connectQueue = new ConcurrentLinkedQueue<ConnectionRequest>();
// 连接请求 取消队列
private final Queue<ConnectionRequest> cancelQueue = new ConcurrentLinkedQueue<ConnectionRequest>();

 
AbstractPollingIoConnector中worker线程的核心程序如下：

while (selectable) {
try {
//最多阻塞1秒
boolean selected = select(1000);

//1 从connectQueue中取出 requst并注册新的内容
nHandles += registerNew();
//2 处理选出的handle socket-channel，推送到processor
if (selected) {
nHandles -= processSessions(selectedHandles());
}
//3 读取所有的handle socket-channel
//对他们进行超时检查
processTimedOutSessions(allHandles());

//4 close掉所有得到的cancel队列中的request
nHandles -= cancelKeys();
//
//如果已经没有handle socket-channel 待处理的话则
//停止本次worker的运行。
//（也就是说connector的worker不是一直运行在那里的）
if (nHandles == 0) {
synchronized (lock) {
if (connectQueue.isEmpty()) {
worker = null;
break;
}
}
}
//
} catch (Throwable e) {
ExceptionMonitor.getInstance().exceptionCaught(e);
//如果出错则休息1秒
try {
Thread.sleep(1000);
} catch (InterruptedException e1) {
ExceptionMonitor.getInstance().exceptionCaught(e1);
}
}

 
与Acceptor类似核心部分就是一个select()方法的循环处理。通过两个Queue进行流转。