mina架构分析 (NIO 网络接口)

出处：http://gearever.iteye.com

Apache MINA(Multipurpose Infrastructure for Network Applications) 是 Apache 组织一个较新的项目，它为开发高性能和高可用性的网络应用程序提供了非常便利的框架。当前发行的 MINA 版本支持基于 Java NIO 技术的 TCP/UDP 应用程序开发、串口通讯程序（只在最新的预览版中提供），MINA 所支持的功能也在进一步的扩展中。
目前正在使用 MINA 的软件包括有：Apache Directory Project、AsyncWeb、AMQP（Advanced Message Queuing Protocol）、RED5 Server（Macromedia Flash Media RTMP）、ObjectRADIUS、Openfire 等等。
---百度百科
本文以mina 2.0.7为基础，记录mina nio网络接口的内部结构及消息流。在tomcat架构分析(connector NIO实现)中描述了tomcat的nio实现，两者可以做一个比较。应该说在一些思想上有共通的地方，在实现上可能mina的非阻塞更彻底一些。

	  public static void main(String[] args) {  
	        SocketAcceptor acceptor = new NioSocketAcceptor();  
	        acceptor.getFilterChain().addLast(  
	                "text",  new Business0Filter();  
	        acceptor.setHandler(new ServerHandler());  
	        try {  
	            acceptor.bind(new InetSocketAddress(PORT));  
	        } catch (IOException e) {  
	       	 acceptor.dispose();  
	        }  
	    }  

这是一个典型的基于mina的服务端程序。Mina的大体结构还是很清晰的，绑定一个端口接收消息，经过一个Filter链(上行)预处理request，最后在Handler完成业务逻辑，然后反向经过Filter链(下行)封装response，直至返回客户端。Mina支持TCP/UDP，本文只记录TCP(面向连接)的消息流。
看一下NioSocketAcceptor的内部结构概念图；

可以看出，整体上还是采用多Selector的架构。

Acceptor线程
在NioSocketAcceptor中有个Acceptor线程，它负责将一个ServerSocketChannel注册到主Selector中，注册事件为OP_ACCEPT，当有OP_ACCEPT响应时，取出相应的socket，封装在NioSocketSession中，并给这个NioSocketSession对象分配一个NioProcessor；

SimpleIoProcessorPool
NioSocketAcceptor维护的一个NioProcessor的对象池，这个对象池缺省状态下维护了Runtime.getRuntime().availableProcessors() + 1个NioProcessor对象，每个运行时建立的NioSocketSession对象都会分配一个NioProcessor对象，这个NioProcessor对象与NioSocketSession对象是一对多的关系；

NioProcessor
每个NioProcessor对象内部维护了一个Selector对象及Processor线程，除此还有一些数据结构，用于数据处理，主要的有；

• Queue<S> newSessions = new ConcurrentLinkedQueue<S>()
• Queue<S> flushingSessions = new ConcurrentLinkedQueue<S>()

其中，newSessions就是当给NioSocketSession对象分配NioProcessor时，将此NioSocketSession对象添加到此newSessions queue中，同时将对应的channel以OP_READ注册到这个NioProcessor对象的Selector，给session分配NioProcessor的逻辑是；

private IoProcessor<S> getProcessor(S session) {
        IoProcessor<S> processor = (IoProcessor<S>) session.getAttribute(PROCESSOR);
        if (processor == null) {
            if (disposed || disposing) {
                throw new IllegalStateException("A disposed processor cannot be accessed.");
            }
            processor = pool[Math.abs((int) session.getId()) % pool.length];
            if (processor == null) {
                throw new IllegalStateException("A disposed processor cannot be accessed.");
            }
            session.setAttributeIfAbsent(PROCESSOR, processor);
        }
        return processor;
}

这个session id是生成session时产生的。

NioProcessor对象的Processor线程就是从这个newSessions queue里取出新加进来的session对象，以OP_READ注册到自己Selector中，然后针对自己的Selector进行select()操作，当有可读的socket事件响应时就取出socket数据，然后走Filter链直至业务逻辑。当有了response需要返回给客户端时，会将response对象封装并添加到session对象的一个写出缓冲里，然后session对象被添加到NioProcessor对象的flushingSessions中，Processor线程会从flushingSessions中取出session，然后从session的写出缓冲中取出response，写socket。需要注意的是，在处理写的时候还是比较细腻的，进入flushingSessions queue有两种方式，

• 一个是经过业务逻辑处理，产生response，直接将session压入flushingSessions queue；
• 另一个是当执行上述写socket动作时，因为网络闪断或其他什么原因，写socket不能正常完成，这时，会将session对应的channel在Selector中注册OP_WRITE事件，当Processor线程发现这个session对应的channel有OP_WRITE事件时，会重新将这个session压入flushingSessions queue，然后就是走正常逻辑了，看下代码感受一下；

private boolean flushNow(S session, long currentTime) {
        final WriteRequestQueue writeRequestQueue = session.getWriteRequestQueue();
        WriteRequest req = null;
        try {
            ......
            do {
                req = session.getCurrentWriteRequest();
                if (req == null) {
                    // session的写出缓冲中取response
                    req = writeRequestQueue.poll(session);
                    if (req == null) {
                        break;
                    }
                    session.setCurrentWriteRequest(req);
                }

                int localWrittenBytes = 0;
                Object message = req.getMessage();
                if (message instanceof IoBuffer) {
                    //写socket
                    localWrittenBytes = writeBuffer(session, req, hasFragmentation, maxWrittenBytes - writtenBytes, currentTime);

                    if ((localWrittenBytes > 0) && ((IoBuffer) message).hasRemaining()) {
                        //写socket成功，但是buffer中还有需要写的数据
                        writtenBytes += localWrittenBytes;
                        setInterestedInWrite(session, true);
                        return false;
                    }
                }
                
                ......
				//写socket出错，重置channel事件
                if (localWrittenBytes == 0) {
                    setInterestedInWrite(session, true);
                    return false;
                }
                ......
				
            } while (writtenBytes < maxWrittenBytes);
        } catch (Exception e) {
            ......
        }
        return true;
    }

    protected void setInterestedInWrite(NioSession session, boolean isInterested) throws Exception {
        SelectionKey key = session.getSelectionKey();
        if (key == null) {
            return;
        }
        int newInterestOps = key.interestOps();
		//注册OP_WRITE事件
        if (isInterested) {
            newInterestOps |= SelectionKey.OP_WRITE;
        } else {
            newInterestOps &= ~SelectionKey.OP_WRITE;
        }
        key.interestOps(newInterestOps);
    }

在各个Selector的维护及session的超时控制等方面，mina在实现的细节上还有很多特别的设计，不一一赘述了。和tomcat的nio框架比起来，感觉mina还是很细腻的，毕竟它是专门的nio框架。很多值得学习的地方。