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

前面介绍过IoSessionRecycler是负责回收不再使用的会话的接口，ExpiringSessionRecycler是其一个实现类，用于回收超时失效的会话。
private ExpiringMap<Object, IoSession> sessionMap;//待处理的会话集
private ExpiringMap<Object, IoSession>.Expirer mapExpirer;//负责具体的回收工作

sessionMap的键是由本地地址和远端地址共同组成的，值是这两个地址对应的会话。

Expirer类实现了Runnable接口，这个线程负责监控ExpiringMap，并把ExpiringMap中超过阀值的元素从ExpiringMap中移除。这个线程调用了setDaemon(true)，因此是作为守护线程在后台运行。具体的处理过程如下：

private void processExpires()
{
long timeNow = System.currentTimeMillis();//当前时间
for (ExpiringObject o : delegate.values())
{
if (timeToLiveMillis <= 0)
{
continue;
}
long timeIdle = timeNow - o.getLastAccessTime();//时间差
if (timeIdle >= timeToLiveMillis)
{//超时
delegate.remove(o.getKey());
for (ExpirationListener<V> listener : expirationListeners)
{//呼叫监听者
listener.expired(o.getValue());
}
}
}
}

启动/关闭超时检查线程都需要进行封锁机制,这里使用的是读写锁:

private final ReadWriteLock stateLock = new ReentrantReadWriteLock();
public void startExpiring()
{
stateLock.writeLock().lock();
try
{
if (!running)
{
running = true;
expirerThread.start();
}
}
finally
{
stateLock.writeLock().unlock();
}
}
public void stopExpiring()
{
stateLock.writeLock().lock();
try
{
if (running)
{
running = false;
expirerThread.interrupt();
}
}
finally
{
stateLock.writeLock().unlock();
}
}

会话超时监听者：

private class DefaultExpirationListener implements
ExpirationListener<IoSession> {
public void expired(IoSession expiredSession) {
expiredSession.close();//关闭超时的会话
}
}

上文的内容还有一些没有结尾，这篇补上。在ExpiringMap类中，使用了一个私有内部类ExpiringObject来表示待检查超时的对象，它包括三个域，键，值，上次访问时间，以及用于上次访问时间这个域的读写锁：

private K key;
private V value;
private long lastAccessTime;
private final ReadWriteLock lastAccessTimeLock = new ReentrantReadWriteLock();

而ExpiringMap中包括了下述几个变量:

private final ConcurrentHashMap<K, ExpiringObject> delegate;//超时代理集合，保存待检查对象
private final CopyOnWriteArrayList<ExpirationListener<V>> expirationListeners;//超时监听者
private final Expirer expirer;//超时检查线程

现在再来看看IoSession的一个抽象实现类AbstractIoSession。这是它的几个重要的成员变量：

private IoSessionAttributeMap attributes;//会话属性映射图
private WriteRequestQueue writeRequestQueue;//写请求队列
private WriteRequest currentWriteRequest;//当前写请求

当要结束当前会话时，会发送一个一个写请求CLOSE_REQUEST。而closeFuture这个CloseFuture会在连接关闭时状态被设置为”closed”,它的监听器是SCHEDULED_COUNTER_RESETTER。

close和closeOnFlush都是异步的关闭操作，区别是前者立即关闭连接，而后者是在写请求队列中放入一个CLOSE_REQUEST，并将其即时刷新出去，若要真正等待关闭完成，需要调用方在返回的CloseFuture等待

public final CloseFuture close() {
synchronized (lock) {
if (isClosing()) {
return closeFuture;
} else {
closing = true;
}
}
getFilterChain().fireFilterClose();//fire出关闭事件
return closeFuture;
}
public final CloseFuture closeOnFlush() {
getWriteRequestQueue().offer(this, CLOSE_REQUEST);
getProcessor().flush(this);
return closeFuture;
}

下面来看看读数据的过程：

public final CloseFuture close() {
synchronized (lock) {
if (isClosing()) {
return closeFuture;
} else {
closing = true;
}
}
getFilterChain().fireFilterClose();//fire出关闭事件
return closeFuture;
}
public final CloseFuture closeOnFlush() {
getWriteRequestQueue().offer(this, CLOSE_REQUEST);
getProcessor().flush(this);
return closeFuture;
}
private Queue<ReadFuture> getReadyReadFutures() {//返回可被读数据队列
Queue<ReadFuture> readyReadFutures =
(Queue<ReadFuture>) getAttribute(READY_READ_FUTURES_KEY);//从会话映射表中取出可被读数据队列
if (readyReadFutures == null) {//第一次读数据
readyReadFutures = new CircularQueue<ReadFuture>();//构造一个新读数据队列
Queue<ReadFuture> oldReadyReadFutures =
(Queue<ReadFuture>) setAttributeIfAbsent(
READY_READ_FUTURES_KEY, readyReadFutures);
if (oldReadyReadFutures != null) {
readyReadFutures = oldReadyReadFutures;
}
}
return readyReadFutures;
}
public final ReadFuture read() {//读数据
if (!getConfig().isUseReadOperation()) {//会话配置不允许读数据(这是默认情况)
throw new IllegalStateException("useReadOperation is not enabled.");
}
Queue<ReadFuture> readyReadFutures = getReadyReadFutures();//获取已经可被读数据队列
ReadFuture future;
synchronized (readyReadFutures) {//锁住读数据队列
future = readyReadFutures.poll();//取队头数据
if (future != null) {
if (future.isClosed()) {//关联的会话已经关闭了，让读者知道此情况
readyReadFutures.offer(future);
}
} else {
future = new DefaultReadFuture(this);
getWaitingReadFutures().offer(future); //将此数据插入等待被读取数据的队列，这个代码和上面的getReadyReadFutures类似，只是键值不同而已
}
}
return future;
}

再来看写数据到指定远端地址的过程，可以写三种类型数据：IoBuffer，整个文件或文件的部分区域，这会通过传递写请求给过滤器链条来完成数据向目的远端的传输。

public final WriteFuture write(Object message, SocketAddress remoteAddress) {
FileChannel openedFileChannel = null;
try
{
if (message instanceof IoBuffer&& !((IoBuffer) message).hasRemaining())
{// 空消息
throw new IllegalArgumentException(
"message is empty. Forgot to call flip()?");
}
else if (message instanceof FileChannel)
{//要发送的是文件的某一区域
FileChannel fileChannel = (FileChannel) message;
message = new DefaultFileRegion(fileChannel, 0, fileChannel.size());
}
else if (message instanceof File)
{//要发送的是文件,打开文件通道
File file = (File) message;
openedFileChannel = new FileInputStream(file).getChannel();
message = new DefaultFileRegion(openedFileChannel, 0, openedFileChannel.size());
}
}
catch (IOException e)
{
ExceptionMonitor.getInstance().exceptionCaught(e);
return DefaultWriteFuture.newNotWrittenFuture(this, e);
}
WriteFuture future = new DefaultWriteFuture(this);
getFilterChain().fireFilterWrite(
new DefaultWriteRequest(message, future, remoteAddress)); //构造写请求,通过过滤器链发送出去，写请求中指明了要发送的消息，目的地址，以及返回的结果

//如果打开了一个文件通道（发送的文件的部分区域或全部），就必须在写请求完成时关闭文件通道
if (openedFileChannel != null) {
final FileChannel finalChannel = openedFileChannel;
future.addListener(new IoFutureListener<WriteFuture>() {
public void operationComplete(WriteFuture future) {
try {
finalChannel.close();//关闭文件通道
} catch (IOException e) {
ExceptionMonitor.getInstance().exceptionCaught(e);
}
}
});
}
return future;//写请求成功完成
}

最后，来看看一个WriteRequestQueue的实现，唯一加入的一个功能就是若在队头发现是请求关闭，则会去关闭会话。

private class CloseRequestAwareWriteRequestQueue implements WriteRequestQueue {
private final WriteRequestQueue q;//内部实际的写请求队列
public CloseRequestAwareWriteRequestQueue(WriteRequestQueue q) {
this.q = q;
}
public synchronized WriteRequest poll(IoSession session) {
WriteRequest answer = q.poll(session);
if (answer == CLOSE_REQUEST) {
AbstractIoSession.this.close();
dispose(session);
answer = null;
}
return answer;
}
public void offer(IoSession session, WriteRequest e) {
q.offer(session, e);
}
public boolean isEmpty(IoSession session) {
return q.isEmpty(session);
}
public void clear(IoSession session) {
q.clear(session);
}
public void dispose(IoSession session) {
q.dispose(session);
}
}

前面介绍完了org.apache.mina.core.session这个包，现在开始进入org.apache.mina.core. polling包。这个包里包含了实现基于轮询策略（比如NIO的select调用或其他类型的I/O轮询系统调用（如epoll,poll,kqueue等）的基类。

先来看AbstractPollingIoAcceptor这个抽象基类，它继承自AbstractIoAcceptor，两个泛型参数分别是所处理的会话和服务器端socket连接。底层的sockets会被不断检测，并当有任何一个socket需要被处理时就会被唤醒去处理。这个类封装了服务器端socket的bind,accept和dispose等动作，其成员变量Executor负责接受来自客户端的连接请求，另一个AbstractPollingIoProcessor用于处理客户端的I/O操作请求，如读写和关闭连接。

其最重要的几个成员变量是：

private final Queue<AcceptorOperationFuture> registerQueue = new ConcurrentLinkedQueue<AcceptorOperationFuture>();//注册队列
private final Queue<AcceptorOperationFuture> cancelQueue = new ConcurrentLinkedQueue<AcceptorOperationFuture>();//取消注册队列
private final Map<SocketAddress, H> boundHandles = Collections
.synchronizedMap(new HashMap<SocketAddress, H>());//本地地址到服务器socket的映射表

先来看看当服务端调用bind后的处理过程：

protected final Set<SocketAddress> bind0(
List<? extends SocketAddress> localAddresses) throws Exception {
AcceptorOperationFuture request = new AcceptorOperationFuture(localAddresses);//注册请求
registerQueue.add(request);//加入注册队列中，等待worker处理
//创建一个Worker实例，开始工作
startupWorker();
wakeup();
request.awaitUninterruptibly();
// 更新本地绑定地址
Set<SocketAddress> newLocalAddresses = new HashSet<SocketAddress>();
for (H handle : boundHandles.values()) {
newLocalAddresses.add(localAddress(handle));
}
return newLocalAddresses;
}

真正的负责接收客户端请求的工作都是Worker线程完成的,

private class Worker implements Runnable {
public void run() {
int nHandles = 0;
while (selectable) {
try {
// Detect if we have some keys ready to be processed
boolean selected = select();//检测是否有SelectionKey已经可以被处理了
nHandles += registerHandles();//注册服务器sockets句柄，这样做的目的是将Selector的状态置于OP_ACCEPT，并绑定到所监听的端口上，表明接受了可以接收的来自客户端的连接请求，
if (selected) {
processHandles(selectedHandles());//处理可以被处理的SelectionKey状态为OP_ACCEPT的服务器socket句柄集(即真正处理来自客户端的连接请求)
}
nHandles -= unregisterHandles();//检查是否有取消连接的客户端请求
if (nHandles == 0) {
synchronized (lock) {
if (registerQueue.isEmpty()
&& cancelQueue.isEmpty()) {//完成工作
worker = null;
break;
}
}
}
} catch (Throwable e) {
ExceptionMonitor.getInstance().exceptionCaught(e);
try {
Thread.sleep(1000);//线程休眠一秒
} catch (InterruptedException e1) {
ExceptionMonitor.getInstance().exceptionCaught(e1);
}
}
}
if (selectable && isDisposing()) {//释放资源
selectable = false;
try {
if (createdProcessor) {
processor.dispose();
}
} finally {
try {
synchronized (disposalLock) {
if (isDisposing()) {
destroy();
}
}
} catch (Exception e) {
ExceptionMonitor.getInstance().exceptionCaught(e);
} finally {
disposalFuture.setDone();
}
}
}
}
private int registerHandles() {//注册服务器sockets句柄
for (;;) {
AcceptorOperationFuture future = registerQueue.poll();
Map<SocketAddress, H> newHandles = new HashMap<SocketAddress, H>();
List<SocketAddress> localAddresses = future.getLocalAddresses();
try {
for (SocketAddress a : localAddresses) {
H handle = open(a);//打开指定地址，返回服务器socket句柄
newHandles.put(localAddress(handle), handle);//加入地址—服务器socket映射表中
}
boundHandles.putAll(newHandles);//更新本地绑定地址集
// and notify.
future.setDone();//完成注册过程
return newHandles.size();
} catch (Exception e) {
future.setException(e);
} finally {
// Roll back if failed to bind all addresses.
if (future.getException() != null) {
for (H handle : newHandles.values()) {
try {
close(handle);//关闭服务器socket句柄
} catch (Exception e) {
ExceptionMonitor.getInstance().exceptionCaught(e);
}
}
wakeup();
}
}
}
}
private void processHandles(Iterator<H> handles) throws Exception {//处理来自客户端的连接请求
while (handles.hasNext()) {
H handle = handles.next();
handles.remove();
T session = accept(processor, handle);//为一个服务器socket句柄handle真正接收来自客户端的请求，在给定的所关联的processor上返回会话session
if (session == null) {
break;
}
finishSessionInitialization(session, null, null);//结束会话初始化
// add the session to the SocketIoProcessor
session.getProcessor().add(session);
}
}
}

这个类中有个地方值得注意，就是wakeup方法，它是用来中断select方法的，当注册队列或取消注册队列发生变化时需要调用它，可以参看本类的一个子类NioSocketAcceptor的实现：

protected boolean select() throws Exception {
return selector.select() > 0;
}
protected void wakeup() {
selector.wakeup();
}

我们可以查阅jdk文档，它对Selector的select方法有如下解释：选择一组键，其相应的通道已为 I/O 操作准备就绪。 此方法执行处于阻塞模式的选择操作。仅在至少选择一个通道、调用此选择器的 wakeup 方法、当前的线程已中断，或者给定的超时期满（以先到者为准）后此方法才返回。