Mina2.0框架源码剖析（一）

整个框架最核心的几个包是：org.apache.mina.core.service,
org.apache.mina.core.session,
org.apache.mina.core.polling以及org.apache.mina.transport.socket。

这一篇先来看org.apache.mina.core.service。第一个要说的接口是IoService,它是所有IoAcceptor和
IoConnector的基接口.对于一个IoService,有哪些信息需要我们关注呢？1）底层的元数据信息TransportMetadata，比
如底层的网络服务提供者（NIO,ARP,RXTX等），2）通过这个服务创建一个新会话时，新会话的默认配置IoSessionConfig。3）此服
务所管理的所有会话。4）与这个服务相关所产生的事件所对应的监听者（IoServiceListener）。5）处理这个服务所管理的所有连接的处理器
(IoHandler)。6）每个会话都有一个过滤器链（IoFilterChain），每个过滤器链通过其对应的
IoFilterChainBuilder来负责构建。7）由于此服务管理了一系列会话，因此可以通过广播的方式向所有会话发送消息,返回结果是一个
WriteFuture集，后者是一种表示未来预期结果的数据结构。8）服务创建的会话（IoSession）相关的数据通过
IoSessionDataStructureFactory来提供。9）发送消息时有一个写缓冲队列。10）服务的闲置状态有三种：读端空闲，写端空
闲，双端空闲。11）还提供服务的一些统计信息，比如时间，数据量等。

IoService这个服务是对于服务器端的接受连接和客户端发起连接这两种行为的抽象。

再来从服务器看起，IoAcceptor是IoService
的子接口，它用于绑定到指定的ip和端口，从而接收来自客户端的连接请求，同时会fire相应的客户端连接成功接收/取消/失败等事件给自己的
IoHandle去处理。当服务器端的Accpetor从早先绑定的ip和端口上取消绑定时，默认是所有的客户端会话会被关闭，这种情况一般出现在服务器
挂掉了，则客户端收到连接关闭的提示。这个接口最重要的两个方法是bind()和unbind()，当这两个方法被调用时，服务端的连接接受线程就启动或
关闭了。

再来看一看客户端的连接发起者接口IoConnector，它的功能和IoAcceptor基本对应的，它用于尝试连接到服务器指定的ip和端口，同时会
fire相应的客户端连接事件给自己的IoHandle去处理。当connet方法被调用后用于连接服务器端的线程就启动了，而当所有的连接尝试都结束时
线程就停止。尝试连接的超时时间可以自行设置。Connect方法返回的结果是ConnectFuture，这和前面说的WriteFuture类似，在
后面会有一篇专门讲这个模式的应用。

前面的IoAcceptor和IoConnector就好比是两个负责握手的仆人，而真正代表会话的实际I/O操作的接口是IoProcessor，它对
现有的Reactor模式架构的Java NIO框架继续做了一层封装。它的泛型参数指明了它能处理的会话类型。接口中最重要的几个方法，
add用于将指定会话加入到此Processor中，
让它负责处理与此会话相关的所有I/O操作。由于写操作会有一个写请求队列，flush就用于对指定会话的写请求队列进行强制刷数据。remove方法用
于从此Processor中移除和关闭指定会话，这样就可以关闭会话相关联的连接并释放所有相关资源。updateTrafficMask方法用于控制会
话的I/O行为，比如是否允许读/写。

然后来说说IoHandle接口，Mina中的所有I/O事件都是通过这个接口来处理的，这些事件都是上面所说的I/O Processor发出来的，要注意的一点是同一个I/O Processor线程是负责处理多个会话的。包括下面这几个事件的处理：

Java代码

public

interface
IoHandler

{

void
sessionCreated(IoSession session)
throws
Exception;
//会话创建

void
sessionOpened(IoSession session)
throws
Exception;
//打开会话,与sessionCreated最大的区别是它是从另一个线程处调用的

void
sessionClosed(IoSession session)
throws
Exception;
//会话结束，当连接关闭时被调用

void
sessionIdle(IoSession session, IdleStatus status)
throws
Exception;
//会话空闲

void
exceptionCaught(IoSession session, Throwable cause)
throws
Exception;
//异常捕获，Mina会自动关闭此连接

void
messageReceived(IoSession session, Object message)
throws
Exception;
//接收到消息

void
messageSent(IoSession session, Object message)
throws
Exception;
//发送消息

}

public interface IoHandler
{
void sessionCreated(IoSession session) throws Exception;//会话创建
void sessionOpened(IoSession session) throws Exception;//打开会话,与sessionCreated最大的区别是它是从另一个线程处调用的
void sessionClosed(IoSession session) throws Exception;//会话结束，当连接关闭时被调用
void sessionIdle(IoSession session, IdleStatus status) throws Exception;//会话空闲
void exceptionCaught(IoSession session, Throwable cause) throws Exception;//异常捕获，Mina会自动关闭此连接
void messageReceived(IoSession session, Object message) throws Exception;//接收到消息
void messageSent(IoSession session, Object message) throws Exception;//发送消息
}

IoHandlerAdapter就不说了，简单地对IoHandler使用适配器模式封装了下，让具体的IoHandler子类从其继承后，从而可以对自身需要哪些事件处理拥有自主权。

来看看IoServiceListener接口，它用于监听IoService相关的事件。

Java代码

public

interface
IoServiceListener
extends
EventListener

{

void
serviceActivated(IoService service)
throws
Exception;
//激活了一个新service

void
serviceIdle(IoService service, IdleStatus idleStatus)
throws
Exception;
// service闲置

void
serviceDeactivated(IoService service)
throws
Exception;
//挂起一个service

void
sessionCreated(IoSession session)
throws
Exception;
//创建一个新会话

void
sessionDestroyed(IoSession session)
throws
Exception;
//摧毁一个新会话

}

public interface IoServiceListener extends EventListener
{
void serviceActivated(IoService service) throws Exception;//激活了一个新service
void serviceIdle(IoService service, IdleStatus idleStatus) throws Exception; // service闲置
void serviceDeactivated(IoService service) throws Exception;//挂起一个service
void sessionCreated(IoSession session) throws Exception;//创建一个新会话
void sessionDestroyed(IoSession session) throws Exception;//摧毁一个新会话
}

IoServiceListenerSupport类就是负责将上面的IoService和其对应的各个IoServiceListener包装到一起进行管理。下面是它的成员变量：

Java代码

private

final
IoService service;

private

final
List<IoServiceListener> listeners =
new
CopyOnWriteArrayList<IoServiceListener>();

private

final
ConcurrentMap<Long, IoSession> managedSessions =
new
ConcurrentHashMap<Long, IoSession>();
//被管理的会话集（其实就是服务所管理的会话集）

private

final
Map<Long, IoSession> readOnlyManagedSessions = Collections.unmodifiableMap(managedSessions);
//上面的会话集的只读版

private

final
AtomicBoolean activated =
new
AtomicBoolean();
//被管理的服务是否处于激活状态

激活事件就以会话创建为例来说明：

public

void
fireSessionCreated(IoSession session)

{

boolean
firstSession =
false
;

if
(session.getService()
instanceof
IoConnector)

{//若服务类型是Connector，则说明是客户端的连接服务

synchronized
(managedSessions)

{//锁住当前已经建立的会话集

firstSession = managedSessions.isEmpty();//看服务所管理的会话集是否为空集

}

}

if
(managedSessions.putIfAbsent(Long.valueOf(session.getId()), session) !=
null
) {
// If already registered, ignore.

return
;

}

if
(firstSession)

{//第一个连接会话，fire一个虚拟的服务激活事件

fireServiceActivated();

}

//呼叫过滤器的事件处理

session.getFilterChain().fireSessionCreated();// 会话创建

session.getFilterChain().fireSessionOpened();//会话打开

int
managedSessionCount = managedSessions.size();

//统计管理的会话数目

if
(managedSessionCount > largestManagedSessionCount)

{

largestManagedSessionCount = managedSessionCount;

}

cumulativeManagedSessionCount ++;

//呼叫监听者的事件处理函数

for
(IoServiceListener l : listeners)

{

try

{

l.sessionCreated(session);

} catch
(Throwable e)

{

ExceptionMonitor.getInstance().exceptionCaught(e);

}

}

}

private final IoService service;
private final List<IoServiceListener> listeners = new CopyOnWriteArrayList<IoServiceListener>();
private final ConcurrentMap<Long, IoSession> managedSessions = new ConcurrentHashMap<Long, IoSession>();//被管理的会话集（其实就是服务所管理的会话集）
private final Map<Long, IoSession> readOnlyManagedSessions = Collections.unmodifiableMap(managedSessions);//上面的会话集的只读版
private final AtomicBoolean activated = new AtomicBoolean();//被管理的服务是否处于激活状态

激活事件就以会话创建为例来说明：
public void fireSessionCreated(IoSession session)
{
boolean firstSession = false;
if (session.getService() instanceof IoConnector)
{//若服务类型是Connector，则说明是客户端的连接服务
synchronized (managedSessions)
{//锁住当前已经建立的会话集
firstSession = managedSessions.isEmpty();//看服务所管理的会话集是否为空集
}
}
if (managedSessions.putIfAbsent(Long.valueOf(session.getId()), session) != null) { // If already registered, ignore.
return;
}
if (firstSession)
{//第一个连接会话，fire一个虚拟的服务激活事件
fireServiceActivated();
}
//呼叫过滤器的事件处理
session.getFilterChain().fireSessionCreated();// 会话创建
session.getFilterChain().fireSessionOpened();//会话打开
int managedSessionCount = managedSessions.size();
//统计管理的会话数目
if (managedSessionCount > largestManagedSessionCount)
{
largestManagedSessionCount = managedSessionCount;
}
cumulativeManagedSessionCount ++;
//呼叫监听者的事件处理函数
for (IoServiceListener l : listeners)
{
try
{
l.sessionCreated(session);
} catch (Throwable e)
{
ExceptionMonitor.getInstance().exceptionCaught(e);
}
}
}

这里值得注意的一个地方是断开连接会话，设置了一个监听锁，直到所有连接会话被关闭后才放开这个锁。

Java代码

private

void
disconnectSessions()

{

if
(!(service
instanceof
IoAcceptor))

{//确保服务类型是IoAcceptor

return
;

}

if
(!((IoAcceptor) service).isCloseOnDeactivation())

{// IoAcceptor是否设置为在服务失效时关闭所有连接会话

return
;

}

Object lock = new
Object();
//监听锁

IoFutureListener<IoFuture> listener = new
LockNotifyingListener(lock);

for
(IoSession s : managedSessions.values())

{

s.close().addListener(listener);//为每个会话的close动作增加一个监听者

}

try

{

synchronized
(lock)

{

while
(!managedSessions.isEmpty())

{//所管理的会话还没有全部结束，持锁等待

lock.wait(500
);

}

}

} catch
(InterruptedException ie)

{

// Ignored

}

}

private

static

class
LockNotifyingListener
implements
IoFutureListener<IoFuture>

{

private

final
Object lock;

public
LockNotifyingListener(Object lock)

{

this
.lock = lock;

}

public

void
operationComplete(IoFuture future)

{

synchronized
(lock)

{

lock.notifyAll();

}

}

}