Mina 源码分析

终于又有时间和精力来写写东西啦，真开心～

这两天闲来没事翻了翻Mina的源码，小有感触，于是决定简单分享一下。

从服务端出发来分析一下吧。Mina有这么几个核心的接口类：

IoAcceptor:  启动server段监听，启动Acceptor线程,处理连接请求同时生成IoSession并放到一个数组里面。

IoProcessor: 启动Processor线程。循环处理上面提到的数组里面的IoSession. 该读读，该写写。

IoSession:    客户端和服务端的会话抽象

IoFilter:         Mina用来处理消息的节点链中的节点

IoHander:     Mina事件监听类。通常也是客户化开发需要扩展的基础接口。

 

闲言少叙，上一张自己画的sequence 图。因为是第一次画这种图，如有不妥请童鞋们指正～

 

 哈哈通过这张图，什么Nio啊，异步什么的就比较能够明显看出来了。

 

 至于NIO,在NioAcceptor里面，对连接请求的处理是通过Selector来实现的。具体可参考java NIO.

 至于异步，可以发现，实际上的读写都是最终通过Processor来实现的。客户端的读写都是放到一个队列里面，

 Processor 不断loop以完成对队列读写的实际操作并且在完成时设置FutureOperation的标志位。

 

 

还有一些细节的东西，比如Acceptor是只有一个线程，而Processor是有多个线程的。默认情况下Processor的线程数是

主机CPU个数+1 。

 

这个的原理是：

 

为了以下描述的清晰性,先贴一下Demo 代码，以方便以下的分析：

public static void main(String[] args) throws Exception {
        SocketAcceptor acceptor = new NioSocketAcceptor();
        acceptor.setReuseAddress( true );
        DefaultIoFilterChainBuilder chain = acceptor.getFilterChain();
        
        // Add SSL filter if SSL is enabled.
        if (USE_SSL) {
            addSSLSupport(chain);
        }

        // Bind
        acceptor.setHandler(new EchoProtocolHandler());
        acceptor.bind(new InetSocketAddress(PORT));

        System.out.println("Listening on port " + PORT);
        
        for (;;) {
            System.out.println("R: " + acceptor.getStatistics().getReadBytesThroughput() + 
                ", W: " + acceptor.getStatistics().getWrittenBytesThroughput());
            Thread.sleep(3000);
        }
    }

 

 

在AbstracePoolingIoAcceptor构造的时候，传入的processor 是一个SimpleIoProcessorPool的实例,这个类的作用就是维护一个默认CPU+1数量的IoProcessor池。入口如下：

protected AbstractPollingIoAcceptor(IoSessionConfig sessionConfig,
            Class<? extends IoProcessor<T>> processorClass) {
        this(sessionConfig, null, new SimpleIoProcessorPool<T>(processorClass),
                true);
    }

 

 

private static final int DEFAULT_SIZE = Runtime.getRuntime().availableProcessors() + 1;
public SimpleIoProcessorPool(Class<? extends IoProcessor<T>> processorType) {
        this(processorType, null, DEFAULT_SIZE);
    }

 

 

    每次调用IoProccesor(在我们的例子里面是SimpleIoProcessorPool)的add方法的时候，会随机从IoProcessor池里面取一个实例然后与要add的session关联。于是实际调用的就是NioProcessor的add方法. 通过这种方式，每个被调用过的IoProcessor都会启动一个自己的Processor线程。

   贴一下为add过程的源码：

 

 

SimpleIoProcessorPool 类
public final void add(T session) {
        getProcessor(session).add(session);
    }

private IoProcessor<T> getProcessor(T session) {
        IoProcessor<T> processor = (IoProcessor<T>) session.getAttribute(PROCESSOR);
        
        if (processor == null) {
            processor = nextProcessor(session);
            session.setAttributeIfAbsent(PROCESSOR, processor);
        }

        return processor;
    }

private IoProcessor<T> nextProcessor(T session) {
        if (disposed) {
            throw new IllegalStateException(
                    "A disposed processor cannot be accessed.");
        }
        
        return pool[Math.abs((int)session.getId()) % pool.length];
    }

 

    通过代码很清晰的能够看到, 实际上的add 调用还是NioProcessor的 add 调用。