NIO之路4--MINA框架源码解析

MINA框架是基于NIO的异步IO框架，上一文已经对MINA的理论及实践做了分析，本文将对于MINA的整体源码实现进行分析。

通过MINA的实际案例可以发现，MINA的IO实现相比于NIO的使用要简单很多，因为不需要关心IO的具体实现，只需要关心具体的IO数据即可。MINA服务端整体步骤一共就四步：

1、创建IoService：初始化IoService，服务端就是创建IoAcceptor对象，客户端就是创建IoConnector对象

2、添加IoFilter：添加IO过滤器，每个IoService内部都有一个IO过滤器链IoFIlterChain，调用addBefore或addLast等方法将IO过滤器添加到过滤器链上

3、设置IoHandler：给IoService设置IO数据处理器IoHandler对象，用来处理器具体的IO业务数据

4、绑定监听端口号：调用IoService的bind方法监听服务端需要监听的端口号，并开启Selector监听客户端连接

一、初始化IoService

服务器创建IoService是直接创建IoService的子接口IoAcceptor，实现类为NioSocketAcceptor，实例化代码如下：

1     /** 创建默认实例*/
2     IoAcceptor acceptor1 = new NioSocketAcceptor();
3 
4     /** 创建指定数量IoProcessor的实例*/
5     IoAcceptor acceptor2 = new NioSocketAcceptor(2);
6 
7     /** 创建指定线程池的实例*/
8     Executor executor = Executors.newFixedThreadPool(4);
9     IoAcceptor acceptor3 = new NioSocketAcceptor(executor, new SimpleIoProcessorPool<>(NioProcessor.class));

虽然NioSocketAcceptor是实现了IoAcceptor接口，但是并不是直接实现的，而是继承了抽象的实现了IoAcceptor接口的父类AbstractPollingIoAcceptor，按默认的构造方法为例，初始化代码如下：

  1 /** 无参构造函数*/
  2     public NioSocketAcceptor() {
  3         /** 调用父类构造函数*/
  4         super(new DefaultSocketSessionConfig(), NioProcessor.class);
  5         ((DefaultSocketSessionConfig) getSessionConfig()).init(this);
  6     }
  7 
  8     /** AbstractPollingIoAcceptor构造函数
  9      * @param sessionConfig:IoSession的全局配置
 10      * @param processorClass:IoProcessor的Class
 11      * */
 12     protected AbstractPollingIoAcceptor(IoSessionConfig sessionConfig, Class<? extends IoProcessor<S>> processorClass) {
 13         /** 调用重载构造函数*/
 14         this(sessionConfig, null, new SimpleIoProcessorPool<S>(processorClass), true, null);
 15     }
 16 
 17     /** AbstractPollingIoAcceptor构造函数
 18      * @param sessionConfig:IoSession的全局配置
 19      * @param executor:线程池
 20      * @param processor:IoProcessor对象
 21      * @param createdProcessor : 是否创建Processor
 22      * @param selectorProvider : SelectorProvider对象,用于创建Selector
 23      * */
 24     private AbstractPollingIoAcceptor(IoSessionConfig sessionConfig, Executor executor, IoProcessor<S> processor,
 25                                       boolean createdProcessor, SelectorProvider selectorProvider) {
 26         /** 1.调用父亲AbstractIoAcceptor的构造函数*/
 27         super(sessionConfig, executor);
 28 
 29         if (processor == null) {
 30             throw new IllegalArgumentException("processor");
 31         }
 32         /** 2.设置属性processor、createdProcessor值*/
 33         this.processor = processor;
 34         this.createdProcessor = createdProcessor;
 35 
 36         try {
 37             /** 3.根据SelectorProvider对象初始化Selector,和NIO的根据SelectorProvider获取Selector对象逻辑一样 */
 38             init(selectorProvider);
 39 
 40             //标记当前IoAcceptor的Selector已经创建完成,可以接收客户端的连接请求了
 41             selectable = true;
 42         } catch (RuntimeException e) {
 43             throw e;
 44         } catch (Exception e) {
 45             throw new RuntimeIoException("Failed to initialize.", e);
 46         } finally {
 47             if (!selectable) {
 48                 try {
 49                     destroy();
 50                 } catch (Exception e) {
 51                     ExceptionMonitor.getInstance().exceptionCaught(e);
 52                 }
 53             }
 54         }
 55     }
 56 
 57     /** AbstractIoAcceptor构造函数*/
 58     protected AbstractIoAcceptor(IoSessionConfig sessionConfig, Executor executor) {
 59         /** 调用父类AbstractIoService构造函数 */
 60         super(sessionConfig, executor);
 61         defaultLocalAddresses.add(null);
 62     }
 63 
 64     /** AbstractIoService构造函数 */
 65     protected AbstractIoService(IoSessionConfig sessionConfig, Executor executor) {
 66         /** 参数校验*/
 67         if (sessionConfig == null) {
 68             throw new IllegalArgumentException("sessionConfig");
 69         }
 70 
 71         if (getTransportMetadata() == null) {
 72             throw new IllegalArgumentException("TransportMetadata");
 73         }
 74 
 75         if (!getTransportMetadata().getSessionConfigType().isAssignableFrom(sessionConfig.getClass())) {
 76             throw new IllegalArgumentException("sessionConfig type: " + sessionConfig.getClass() + " (expected: "
 77                     + getTransportMetadata().getSessionConfigType() + ")");
 78         }
 79 
 80         /** 创建IoServiceListenerSupport对象,IoServiceListenerSupport内部有一个IoServiceListener的列表
 81          *  将当前的IoServerListener对象添加到IoServiceListenerSupport的列表中
 82          *  */
 83         listeners = new IoServiceListenerSupport(this);
 84         listeners.add(serviceActivationListener);
 85 
 86         /** 设置属性 sessionConfig*/
 87         this.sessionConfig = sessionConfig;
 88 
 89         //加载异常监听器ExceptionMonitor对象,提前加载防止在使用的时候还没有初始化
 90         ExceptionMonitor.getInstance();
 91         /** 设置线程池,如果没有自定义,就采用默认的newCachedThreadPool*/
 92         if (executor == null) {
 93             this.executor = Executors.newCachedThreadPool();
 94             createdExecutor = true;
 95         } else {
 96             this.executor = executor;
 97             createdExecutor = false;
 98         }
 99 
100         threadName = getClass().getSimpleName() + '-' + id.incrementAndGet();
101     }

整个的初始化过程都是在给IoAcceptor的一些属性进行初始化，核心属性包括sessionConfig、executor、IoProcessor等

其中SessionConfig表示IoSession的全局属性配置，主要配置如下：

  1 public interface IoSessionConfig {
  2 
  3         /**
  4          * 获取IoProcessor读取数据的缓冲区大小
  5          */
  6         int getReadBufferSize();
  7 
  8         /**
  9          * 设置IoProcessor读取数据的缓冲区大小,通常会由IoProcessor自动动态调整
 10          */
 11         void setReadBufferSize(int readBufferSize);
 12 
 13         /**
 14          * 获取IoProcessor读取属性的缓冲区大小的最小值
 15          */
 16         int getMinReadBufferSize();
 17 
 18         /**
 19          * 设置IoProcessor读取属性的缓冲区大小的最小值
 20          */
 21         void setMinReadBufferSize(int minReadBufferSize);
 22 
 23         /**
 24          * 获取IoProcessor读取属性的缓冲区大小的最大值
 25          */
 26         int getMaxReadBufferSize();
 27 
 28         /**
 29          * 设置IoProcessor读取属性的缓冲区大小的最大值
 30          */
 31         void setMaxReadBufferSize(int maxReadBufferSize);
 32 
 33         /**
 34          * 获取吞吐量(TPS)的统计间隔,单位为秒,默认是3秒
 35          */
 36         int getThroughputCalculationInterval();
 37 
 38         /**
 39          * 获取吞吐量(TPS)的统计间隔,单位为毫秒
 40          */
 41         long getThroughputCalculationIntervalInMillis();
 42 
 43         /**
 44          * 设置吞吐量的统计间隔时间
 45          */
 46         void setThroughputCalculationInterval(int throughputCalculationInterval);
 47 
 48         /**
 49          * 获取指定状态的空闲时间,不同状态时间可能设置的不一样, 类型如下:
 50          * READER_IDLE : 读数据空闲
 51          * WRITER_IDLE : 写数据空闲
 52          * BOTH_IDLE : 读写都空闲
 53          */
 54         int getIdleTime(IdleStatus status);
 55 
 56         /**
 57          * 获取指定状态的空闲时间,单位为毫秒
 58          */
 59         long getIdleTimeInMillis(IdleStatus status);
 60 
 61         /**
 62          * 设置指定状态的空闲时间
 63          */
 64         void setIdleTime(IdleStatus status, int idleTime);
 65 
 66         /**
 67          * 获取读空闲时间,单位秒
 68          */
 69         int getReaderIdleTime();
 70 
 71         /**
 72          * 获取读空闲时间,单位毫秒
 73          */
 74         long getReaderIdleTimeInMillis();
 75 
 76         /**
 77          * 设置读空闲时间
 78          */
 79         void setReaderIdleTime(int idleTime);
 80 
 81         /**
 82          * 获取写空闲时间,单位秒
 83          */
 84         int getWriterIdleTime();
 85 
 86         /**
 87          * 获取写空闲时间,单位毫秒
 88          */
 89         long getWriterIdleTimeInMillis();
 90 
 91         /**
 92          * 设置写空闲时间
 93          */
 94         void setWriterIdleTime(int idleTime);
 95 
 96         /**
 97          * 获取读写都空闲时间,单位秒
 98          */
 99         int getBothIdleTime();
100 
101         /**
102          * 获取读写都空闲时间,单位毫秒
103          */
104         long getBothIdleTimeInMillis();
105 
106         /**
107          * 设置读写都空闲时间
108          */
109         void setBothIdleTime(int idleTime);
110 
111         /**
112          * 获取写超时时间,单位为秒
113          */
114         int getWriteTimeout();
115 
116         /**
117          * 获取写超时时间,单位毫秒
118          */
119         long getWriteTimeoutInMillis();
120 
121         /**
122          * 设置写超时时间
123          */
124         void setWriteTimeout(int writeTimeout);
125 
126         /**
127          * 获取会话读操作是否开启
128          */
129         boolean isUseReadOperation();
130 
131         /**
132          * 开启或关闭会话读操作,如果开启所有接收到的消息会存储在内存的BlockingQueue中,使客户端应用可以更方便读取接收的消息
133          * 开启这个选项对服务器应用无效,并可能会导致内存泄漏,默认为关闭状态
134          */
135         void setUseReadOperation(boolean useReadOperation);
136 
137         /** 全量设置为IoSessionConfig */
138         void setAll(IoSessionConfig config);
139     }

另外一个属性是线程池对象executor，如果没有自定义线程池传入的话，那么默认会调用Executors.newCachedThreadPool()创建无线程数上限的线程池，所以推荐使用自定义的线程池，避免默认的线程池没有线程数量限制导致线程过多的问题。

还有一个属性是IoProcessor池，默认类为SimpleIoProcessorPool，SimpleIoProcessorPool内部有一个IoProcessor[] pool和一个Executor executor属性，pool是IoProcessor数组，存储所有的IoProcessor，数量可以自定义通过构造函数传入，默认的个数为当前服务器机器的CPU个数+1，比如4核CPU那么就会创建5个IoProcessor，另外每一个IoProcessor初始化的时候都会设置线程池executor属性，如果executor没有自定义同样也会使用Executors.newCachedThreadPool创建。

总结：

IoService初始化的时候涉及到了两个线程池，首先是IoService本身使用的线程池，IoService本身用于接收客户端的连接请求，而连接请求的处理就交给了线程池处理；

另外IoService初始化时会创建多个IoProcessor用于处理客户端具体的IO操作，每一个IoProcessor内部有一个Selector用于监听客户端IO事件，然后将IO事件交给内部的线程池来处理

二、添加IoFilter

IoService内部都一个过滤器链IoFilterChainBuilder对象，默认实现类为DefaultIoFilterChainBuilder类，添加IoFilter时主要有四个方法，分别是在过滤器链的不同位置插入指定过滤器，用法分别如下：

 1         IoAcceptor acceptor = new NioSocketAcceptor();
 2         /** 获取过滤器链 */
 3         DefaultIoFilterChainBuilder ioFilterChain = acceptor.getFilterChain();
 4 
 5         /** 1.在过滤器链的首部加入过滤器 */
 6         ioFilterChain.addFirst("codec", new ProtocolCodecFilter(new TextLineCodecFactory(Charset.forName("UTF-8"))));
 7         /** 2.在过滤器链的指定过滤器前面加入过滤器 */
 8         ioFilterChain.addBefore("logFilter","codec", new ProtocolCodecFilter(new TextLineCodecFactory(Charset.forName("UTF-8"))));
 9         /** 3.在过滤器链的指定过滤器后面加入过滤器 */
10         ioFilterChain.addAfter("logFilter","codec", new ProtocolCodecFilter(new TextLineCodecFactory(Charset.forName("UTF-8"))));
11         /** 4.在过滤器链的尾部加入过滤器 */
12         ioFilterChain.addLast("codec", new ProtocolCodecFilter(new TextLineCodecFactory(Charset.forName("UTF-8"))));

以addFirst为例，实现逻辑如下：

 1 /** 在过滤器链首部插入过滤器
 2      * @param name:过滤器的名称
 3      * @param filter: 过滤器
 4      * */
 5     public synchronized void addFirst(String name, IoFilter filter) {
 6         /** 构造过滤器链路节点对象EntryImpl,并调用register方法加入到列表中 */
 7         register(0, new EntryImpl(name, filter));
 8     }
 9 
10 
11     private final List<Entry> entries;
12 
13     public DefaultIoFilterChainBuilder() {
14         entries = new CopyOnWriteArrayList<Entry>();
15     }
16 
17     /**
18      * 在指定位置插入Entry节点
19      * */
20     private void register(int index, Entry e) {
21         if (contains(e.getName())) {
22             throw new IllegalArgumentException("Other filter is using the same name: " + e.getName());
23         }
24         //调用List的add方法在指定为止插入节点
25         entries.add(index, e);
26     }

从源码可以看出添加IoFilter的逻辑比较简单，过滤器链IoFilterChainBuilder对象内部有一个列表，用于存放所有的IoFilter，添加过滤器就是将IoFilter和名称封装成Entry对象添加到列表中，不同的add方法实际就是调用List的add(int index, Entry e)方法在指定的位置插入元素。addFirst就是在list头部插入元素，addLast就是在list尾部插入元素，addBefore和addAfter先通过遍历查找指定过滤器在List中的位置，然后再将新插入的过滤器插入到指定位置

三、设置IoHandler

 1 public final void setHandler(IoHandler handler) {
 2         if (handler == null) {
 3             throw new IllegalArgumentException("handler cannot be null");
 4         }
 5         /** 判断IoService是否活跃,主要是判断IoServiceListenerSupport是否有活跃的IoServiceListener */
 6         if (isActive()) {
 7             throw new IllegalStateException("handler cannot be set while the service is active.");
 8         }
 9         /** 设置IoHandler属性*/
10         this.handler = handler;
11     }

设置IoHandler的逻辑比较简单，就是给IoService内部的handler属性赋值

四、绑定主机

前面三个步骤都是在初始化服务器并设置各种属性，接下来就是绑定监听的逻辑，源码如下：

 1 public final void bind(SocketAddress localAddress) throws IOException {
 2         if (localAddress == null) {
 3             throw new IllegalArgumentException("localAddress");
 4         }
 5         /** 创建地址列表,将传入的SocketAddress添加到列表中*/
 6         List<SocketAddress> localAddresses = new ArrayList<>(1);
 7         localAddresses.add(localAddress);
 8         /** 内部方法执行绑定逻辑*/
 9         bind(localAddresses);
10     }

调用内部重载的bind方法，源码如下：

 1 public final void bind(Iterable<? extends SocketAddress> localAddresses) throws IOException {
 2         /** 参数校验*/
 3         if (isDisposing()) {
 4             throw new IllegalStateException("The Accpetor disposed is being disposed.");
 5         }
 6 
 7         if (localAddresses == null) {
 8             throw new IllegalArgumentException("localAddresses");
 9         }
10 
11         List<SocketAddress> localAddressesCopy = new ArrayList<>();
12 
13         for (SocketAddress a : localAddresses) {
14             checkAddressType(a);
15             localAddressesCopy.add(a);
16         }
17 
18         if (localAddressesCopy.isEmpty()) {
19             throw new IllegalArgumentException("localAddresses is empty.");
20         }
21 
22         boolean activate = false;
23         synchronized (bindLock) {
24             synchronized (boundAddresses) {
25                 if (boundAddresses.isEmpty()) {
26                     activate = true;
27                 }
28             }
29             /** 判断IoHandler是否为空*/
30             if (getHandler() == null) {
31                 throw new IllegalStateException("handler is not set.");
32             }
33 
34             try {
35                 /**  绑定主机地址 */
36                 Set<SocketAddress> addresses = bindInternal(localAddressesCopy);
37                 synchronized (boundAddresses) {
38                     boundAddresses.addAll(addresses);
39                 }
40             } catch (IOException e) {
41                 throw e;
42             } catch (RuntimeException e) {
43                 throw e;
44             } catch (Exception e) {
45                 throw new RuntimeIoException("Failed to bind to: " + getLocalAddresses(), e);
46             }
47         }
48         /** 激活IoServiceListener */
49         if (activate) {
50             getListeners().fireServiceActivated();
51         }
52     }

虽然代码比较多，但是核心代码就只有 bindInternal这一行， bindInternal是真正的绑定主机的方法，代码如下：

 1 /** 绑定主机*/
 2     protected final Set<SocketAddress> bindInternal(List<? extends SocketAddress> localAddresses) throws Exception {
 3         /** 创建一个Future对象,当IoAcceptor的Selector处理了该请求后会给Future对象发送一个信号 */
 4         AcceptorOperationFuture request = new AcceptorOperationFuture(localAddresses);
 5 
 6         /** 添加到AcceptorOperationFuture队列中*/
 7         registerQueue.add(request);
 8 
 9         /** 开启Acceptor*/
10         startupAcceptor();
11 
12         try {
13             /** 信号量设置为1,相当于加锁处理*/
14             lock.acquire();
15 
16             wakeup();
17         } finally {
18             /** 信号量释放, 相当于解锁处理*/
19             lock.release();
20         }
21 
22         // Now, we wait until this request is completed.
23         request.awaitUninterruptibly();
24 
25         if (request.getException() != null) {
26             throw request.getException();
27         }
28         Set<SocketAddress> newLocalAddresses = new HashSet<>();
29 
30         for (H handle : boundHandles.values()) {
31             newLocalAddresses.add(localAddress(handle));
32         }
33 
34         return newLocalAddresses;
35     }

核心逻辑是调用了startupAcceptor方法，代码如下：

 1 private void startupAcceptor() throws InterruptedException {
 2         if (!selectable) {
 3             registerQueue.clear();
 4             cancelQueue.clear();
 5         }
 6 
 7         /** 创建Acceptor对象, Acceptor实现了Runnable接口*/
 8         Acceptor acceptor = acceptorRef.get();
 9 
10         if (acceptor == null) {
11             lock.acquire();
12             acceptor = new Acceptor();
13 
14             if (acceptorRef.compareAndSet(null, acceptor)) {
15                 /** 将实现了Runnable接口的acceptor对象放入线程池中执行*/
16                 executeWorker(acceptor);
17             } else {
18                 lock.release();
19             }
20         }
21     }

 1 protected final void executeWorker(Runnable worker) {
 2         executeWorker(worker, null);
 3     }
 4 
 5     protected final void executeWorker(Runnable worker, String suffix) {
 6         String actualThreadName = threadName;
 7         if (suffix != null) {
 8             actualThreadName = actualThreadName + '-' + suffix;
 9         }
10         executor.execute(new NamePreservingRunnable(worker, actualThreadName));
11     }

该方法的核心逻辑是创建一个Acceptor对象，Acceptor对象是实现了Runnable接口的，所以是一个可执行逻辑。然后调用executeWorker方法将Acceptor对象交给线程池执行，而线程池就是初始化IoService时初始化的线程池，默认是Executors.newCachedThreadPool

既然Acceptor实现了Runnable接口，那么就再看下Acceptor的具体实现逻辑：

  1 private class Acceptor implements Runnable {
  2         public void run() {
  3             assert (acceptorRef.get() == this);
  4 
  5             int nHandles = 0;
  6 
  7             /** 释放锁*/
  8             lock.release();
  9 
 10             while (selectable) {
 11                 try {
 12                     /** 获取注册的监听端口的数量*/
 13                     nHandles += registerHandles();
 14 
 15                     /** 调用Selector的select()方法,会阻塞当前线程 */
 16                     int selected = select();
 17 
 18                     // Now, if the number of registred handles is 0, we can
 19                     // quit the loop: we don't have any socket listening
 20                     // for incoming connection.
 21                     if (nHandles == 0) {
 22                         acceptorRef.set(null);
 23 
 24                         if (registerQueue.isEmpty() && cancelQueue.isEmpty()) {
 25                             assert (acceptorRef.get() != this);
 26                             break;
 27                         }
 28 
 29                         if (!acceptorRef.compareAndSet(null, this)) {
 30                             assert (acceptorRef.get() != this);
 31                             break;
 32                         }
 33 
 34                         assert (acceptorRef.get() == this);
 35                     }
 36 
 37                     if (selected > 0) {
 38                         /** 处理Selector返回的所有的SelectionKey */
 39                         processHandles(selectedHandles());
 40                     }
 41 
 42                     // check to see if any cancellation request has been made.
 43                     nHandles -= unregisterHandles();
 44                 } catch (ClosedSelectorException cse) {
 45                     // If the selector has been closed, we can exit the loop
 46                     ExceptionMonitor.getInstance().exceptionCaught(cse);
 47                     break;
 48                 } catch (Exception e) {
 49                     ExceptionMonitor.getInstance().exceptionCaught(e);
 50 
 51                     try {
 52                         Thread.sleep(1000);
 53                     } catch (InterruptedException e1) {
 54                         ExceptionMonitor.getInstance().exceptionCaught(e1);
 55                     }
 56                 }
 57             }
 58 
 59             /** 当销毁IoService时, 销毁所有的processor*/
 60             if (selectable && isDisposing()) {
 61                 selectable = false;
 62                 try {
 63                     if (createdProcessor) {
 64                         processor.dispose();
 65                     }
 66                 } finally {
 67                     try {
 68                         synchronized (disposalLock) {
 69                             if (isDisposing()) {
 70                                 destroy();
 71                             }
 72                         }
 73                     } catch (Exception e) {
 74                         ExceptionMonitor.getInstance().exceptionCaught(e);
 75                     } finally {
 76                         disposalFuture.setDone();
 77                     }
 78                 }
 79             }
 80         }
 81 
 82         /**
 83          * This method will process new sessions for the Worker class.  All
 84          * keys that have had their status updates as per the Selector.selectedKeys()
 85          * method will be processed here.  Only keys that are ready to accept
 86          * connections are handled here.
 87          * <p/>
 88          * Session objects are created by making new instances of SocketSessionImpl
 89          * and passing the session object to the SocketIoProcessor class.
 90          */
 91         @SuppressWarnings("unchecked")
 92         private void processHandles(Iterator<H> handles) throws Exception {
 93             /** 遍历所有SelectionKey*/
 94             while (handles.hasNext()) {
 95                 H handle = handles.next();
 96                 handles.remove();
 97 
 98                 /** 处理客户端连接请求,封装成NioSocketSession对象*/
 99                 S session = accept(processor, handle);
100 
101                 if (session == null) {
102                     continue;
103                 }
104                 /** 初始化客户端Session对象, 设置上一次读写时间 */
105                 initSession(session, null, null);
106 
107                 /** 将客户端Session添加到IoProcessor线程池,并分配一个IoProcessor和Session进行绑定 */
108                 session.getProcessor().add(session);
109             }
110         }
111     }
112 
113     protected NioSession accept(IoProcessor<NioSession> processor, ServerSocketChannel handle) throws Exception {
114 
115         SelectionKey key = null;
116 
117         if (handle != null) {
118             key = handle.keyFor(selector);
119         }
120         /** key有效且必须触发了OP_ACCEPT事件,其他事件不处理*/
121         if ((key == null) || (!key.isValid()) || (!key.isAcceptable())) {
122             return null;
123         }
124 
125         /** 获取客户端连接的Channel*/
126         SocketChannel ch = handle.accept();
127 
128         if (ch == null) {
129             return null;
130         }
131         /** 封装客户端连接为NioSocketSession对象 */
132         return new NioSocketSession(this, processor, ch);
133     }
134 
135     public NioSocketSession(IoService service, IoProcessor<NioSession> processor, SocketChannel channel) {
136         /** 调用父类构造函数 设置IoProcessor、IoService、SocketChannel属性*/
137         super(processor, service, channel);
138         config = new SessionConfigImpl();
139         /** 将IoService的全局SessionConfig复制给当前session的配置*/
140         config.setAll(service.getSessionConfig());
141     }

从代码中可以看出IoAcceptor的run方法基本上和NIO的服务器启动方法差不多，首先了调用Selector的select()方法获取所有客户端的请求连接accept事件，然后遍历所有的SelectionKey，将客户端的连接封装成NioSocketSession，最后再将NioSocketSession添加到IoProcessor线程池中，而绑定到具体的IoProcessor的逻辑是在IoProcessor的add方法中实现的，这里IoProcessor线程池是通过SimpleIoProcessorPool类实现的，源码如下：

 1 /** 添加SocketSession*/
 2     public final void add(S session) {
 3         /** 根据session获取具体的IoProcessor,并将session添加到IoProcessor中*/
 4         getProcessor(session).add(session);
 5     }
 6 
 7     private IoProcessor<S> getProcessor(S session) {
 8         IoProcessor<S> processor = (IoProcessor<S>) session.getAttribute(PROCESSOR);
 9 
10         if (processor == null) {
11             if (disposed || disposing) {
12                 throw new IllegalStateException("A disposed processor cannot be accessed.");
13             }
14             /** 根据session的ID通过取模算法从数组中获取对应的IoProcessor*/
15             processor = pool[Math.abs((int) session.getId()) % pool.length];
16 
17             if (processor == null) {
18                 throw new IllegalStateException("A disposed processor cannot be accessed.");
19             }
20             /** 给session设置IoProcessor属性*/
21             session.setAttributeIfAbsent(PROCESSOR, processor);
22         }
23 
24         return processor;
25     }

每一个SocketSession会绑定一个IoProcessor，并会缓存在session的attribute中，而获取IoProcessor的方式则是通过取模算法从IoProcessor数组中获取。

然后调用IoProcessor的add方法将session添加到IoProcessor中，处理IO数据的IoProcessor实现类为NioProcessor，实现的add方法源码如下：

 1 /** 新创建的session队列*/
 2     private final Queue<S> newSessions = new ConcurrentLinkedQueue<>();
 3 
 4     /** 需要移除的session队列 */
 5     private final Queue<S> removingSessions = new ConcurrentLinkedQueue<>();
 6 
 7     /** 需要刷新的session队列*/
 8     private final Queue<S> flushingSessions = new ConcurrentLinkedQueue<>();
 9 
10     public final void add(S session) {
11         if (disposed || disposing) {
12             throw new IllegalStateException("Already disposed.");
13         }
14 
15         /** 将新的session加入到队列中*/
16         newSessions.add(session);
17         /** 开启IoProcessor*/
18         startupProcessor();
19     }
20 
21     private void startupProcessor() {
22         /** 原子引用类型,如果不存在就新建Processor*/
23         Processor processor = processorRef.get();
24 
25         if (processor == null) {
26             processor = new Processor();
27 
28             if (processorRef.compareAndSet(null, processor)) {
29                 /** 将可执行的Processor添加到线程池中执行 */
30                 executor.execute(new NamePreservingRunnable(processor, threadName));
31             }
32         }
33 
34         /** 唤醒Selector, 防止一直被阻塞*/
35         wakeup();
36     }

这里将session添加到队列中，然后开启IoProcessor线程，过程和IoSession类型，都是封装成一个NamePreservingRunnable对象交给线程池去执行，而这里的线程池和IoService的线程池是独立的，没给IoProcessor都有一个自己的线程池

Processor具体的执行逻辑如下：

  1 /** Processor线程,用于处理IO读写操作*/
  2     private class Processor implements Runnable {
  3         public void run() {
  4             assert (processorRef.get() == this);
  5             //会话数量
  6             int nSessions = 0;
  7             lastIdleCheckTime = System.currentTimeMillis();
  8             /** 重试次数*/
  9             int nbTries = 10;
 10 
 11             for (;;) {
 12                 try {
 13                     long t0 = System.currentTimeMillis();
 14                     int selected = select(SELECT_TIMEOUT);
 15                     long t1 = System.currentTimeMillis();
 16                     /** 计算select()方法执行的时间*/
 17                     long delta = t1 - t0;
 18 
 19                     if (!wakeupCalled.getAndSet(false) && (selected == 0) && (delta < 100)) {
 20                         // Last chance : the select() may have been
 21                         // interrupted because we have had an closed channel.
 22                         if (isBrokenConnection()) {
 23                             LOG.warn("Broken connection");
 24                         } else {
 25                             // Ok, we are hit by the nasty epoll
 26                             // spinning.
 27                             // Basically, there is a race condition
 28                             // which causes a closing file descriptor not to be
 29                             // considered as available as a selected channel,
 30                             // but
 31                             // it stopped the select. The next time we will
 32                             // call select(), it will exit immediately for the
 33                             // same
 34                             // reason, and do so forever, consuming 100%
 35                             // CPU.
 36                             // We have to destroy the selector, and
 37                             // register all the socket on a new one.
 38                             if (nbTries == 0) {
 39                                 LOG.warn("Create a new selector. Selected is 0, delta = " + delta);
 40                                 registerNewSelector();
 41                                 nbTries = 10;
 42                             } else {
 43                                 nbTries--;
 44                             }
 45                         }
 46                     } else {
 47                         nbTries = 10;
 48                     }
 49 
 50                     /** 处理新添加的Session
 51                      *  将session对应的channel注册到当前IoProcessor的Selector上,并监听OP_READ事件
 52                      * */
 53                     nSessions += handleNewSessions();
 54 
 55                     updateTrafficMask();
 56 
 57                     // Now, if we have had some incoming or outgoing events,
 58                     // deal with them
 59                     if (selected > 0) {
 60                         /** 当select()返回值大于0表示当前已经有IO事件需要处理,则执行process方法进行处理*/
 61                         process();
 62                     }
 63 
 64                     // Write the pending requests
 65                     long currentTime = System.currentTimeMillis();
 66                     flush(currentTime);
 67 
 68                     /**
 69                      * 移除所有需要移除的session,移除之前会将所有需要发送的数据发送给客户端
 70                      * */
 71                     nSessions -= removeSessions();
 72 
 73                     /** 更新所有session的读空闲、写空闲和读写空闲的时间值 */
 74                     notifyIdleSessions(currentTime);
 75 
 76                     // Get a chance to exit the infinite loop if there are no
 77                     // more sessions on this Processor
 78                     if (nSessions == 0) {
 79                         processorRef.set(null);
 80 
 81                         if (newSessions.isEmpty() && isSelectorEmpty()) {
 82                             // newSessions.add() precedes startupProcessor
 83                             assert (processorRef.get() != this);
 84                             break;
 85                         }
 86 
 87                         assert (processorRef.get() != this);
 88 
 89                         if (!processorRef.compareAndSet(null, this)) {
 90                             // startupProcessor won race, so must exit processor
 91                             assert (processorRef.get() != this);
 92                             break;
 93                         }
 94 
 95                         assert (processorRef.get() == this);
 96                     }
 97 
 98                     /** 当IoProcessor销毁了,则移除所有的客户端的SocketSession*/
 99                     if (isDisposing()) {
100                         boolean hasKeys = false;
101 
102                         for (Iterator<S> i = allSessions(); i.hasNext();) {
103                             IoSession session = i.next();
104 
105                             if (session.isActive()) {
106                                 scheduleRemove((S)session);
107                                 hasKeys = true;
108                             }
109                         }
110 
111                         if (hasKeys) {
112                             wakeup();
113                         }
114                     }
115                 } catch (ClosedSelectorException cse) {
116                     // If the selector has been closed, we can exit the loop
117                     // But first, dump a stack trace
118                     ExceptionMonitor.getInstance().exceptionCaught(cse);
119                     break;
120                 } catch (Exception e) {
121                     ExceptionMonitor.getInstance().exceptionCaught(e);
122 
123                     try {
124                         Thread.sleep(1000);
125                     } catch (InterruptedException e1) {
126                         ExceptionMonitor.getInstance().exceptionCaught(e1);
127                     }
128                 }
129             }
130 
131             try {
132                 synchronized (disposalLock) {
133                     if (disposing) {
134                         doDispose();
135                     }
136                 }
137             } catch (Exception e) {
138                 ExceptionMonitor.getInstance().exceptionCaught(e);
139             } finally {
140                 disposalFuture.setValue(true);
141             }
142         }
143     }

这里代码比较多，但是核心的不多，主要功能如下：

1、处理所有新加入的Session，将session对应的channel注册到Selector中，并监听OP_READ事件

2、处理所有的需要移除的session，将session从Selector中移除监听

3、调用Selector的select()方法获取IO事件，如果存在IO事件，则调用process()方法进行处理

4、更新所有session的读写空闲时间

而处理IO事件的逻辑全部在process方法中处理，源码如下：

 1 /** 处理IoProcessor的Selector监听返回的所有IO事件 */
 2     private void process() throws Exception {
 3         /** 遍历所有的SelectionKey*/
 4         for (Iterator<S> i = selectedSessions(); i.hasNext();) {
 5             S session = i.next();
 6             /** 处理单个客户端session的IO事件*/
 7             process(session);
 8             i.remove();
 9         }
10     }
11     /** 获取所有的SelectionKey*/
12     protected Iterator<NioSession> selectedSessions() {
13         return new IoSessionIterator(selector.selectedKeys());
14     }

process方法的逻辑不多，先是获取selector所有的IO事件SelectionKey集合，然后进行遍历所有的SelectionKey集合，调用process(session)方法处理每个Session的IO事件，源码如下：

 1 /***/
 2     private void process(S session) {
 3         /** 处理读事件 */
 4         if (isReadable(session) && !session.isReadSuspended()) {
 5             /** 处理session的读事件*/
 6             read(session);
 7         }
 8 
 9         /** 处理写事件 */
10         if (isWritable(session) && !session.isWriteSuspended() && session.setScheduledForFlush(true)) {
11             /** 刷新session */
12             flushingSessions.add(session);
13         }
14     }
15 
16     private void read(S session) {
17         /** 获取配置的读缓冲区大小*/
18         IoSessionConfig config = session.getConfig();
19         int bufferSize = config.getReadBufferSize();
20         /** 分配指定大小的缓冲区*/
21         IoBuffer buf = IoBuffer.allocate(bufferSize);
22 
23         final boolean hasFragmentation = session.getTransportMetadata().hasFragmentation();
24 
25         try {
26             int readBytes = 0;
27             int ret;
28 
29             try {
30                 if (hasFragmentation) {
31                     /** 调用read方法将session中的IO数据读取到缓冲区*/
32                     while ((ret = read(session, buf)) > 0) {
33                         readBytes += ret;
34 
35                         if (!buf.hasRemaining()) {
36                             break;
37                         }
38                     }
39                 } else {
40                     /** 调用read方法将session中的IO数据读取到缓冲区*/
41                     ret = read(session, buf);
42                     if (ret > 0) {
43                         readBytes = ret;
44                     }
45                 }
46             } finally {
47                 /** 从读模式切换到写模式 */
48                 buf.flip();
49             }
50 
51             if (readBytes > 0) {
52                 IoFilterChain filterChain = session.getFilterChain();
53                 /** 将读到的数据交给过滤器链进行过滤处理 */
54                 filterChain.fireMessageReceived(buf);
55                 buf = null;
56 
57                 if (hasFragmentation) {
58                     /** 判断读取到的数据是否小于缓冲区总大小的一半*/
59                     if (readBytes << 1 < config.getReadBufferSize()) {
60                         /** 缩小缓冲区大小 */
61                         session.decreaseReadBufferSize();
62                     } else if (readBytes == config.getReadBufferSize()) {
63                         /** 扩大缓冲区大小 */
64                         session.increaseReadBufferSize();
65                     }
66                 }
67             }
68 
69             if (ret < 0) {
70                 /** 过滤器链关闭 */
71                 IoFilterChain filterChain = session.getFilterChain();
72                 filterChain.fireInputClosed();
73             }
74         } catch (Exception e) {
75             if (e instanceof IOException) {
76                 if (!(e instanceof PortUnreachableException)
77                         || !AbstractDatagramSessionConfig.class.isAssignableFrom(config.getClass())
78                         || ((AbstractDatagramSessionConfig) config).isCloseOnPortUnreachable()) {
79                     scheduleRemove(session);
80                 }
81             }
82             /** 如果抛异常,则通知过滤器链执行异常事件*/
83             IoFilterChain filterChain = session.getFilterChain();
84             filterChain.fireExceptionCaught(e);
85         }
86     }

这里逻辑比较多，但是整体逻辑不复杂，主要逻辑如下：

1、如果是可读事件，那么就执行read方法进行IO数据的读取

2、根据配置的读缓冲区大小创建IoBuffer对象进行内存分配

3、调用read方法从session中读取数据存到IoBuffer中，读取完成将IoBuffer从读模式切换到写模式

4、将读取到的IoBuffer数据交给过滤器链进行所有IO过滤器进行过滤处理

5、将缓冲区大小进行扩容或降容处理

6、将连接关闭或连接异常的事件交给过滤器链进行对应的处理

接下来就针对每一个步骤分别进行源码分析

1、分配缓冲区

 1 /** 内存分配工具 */
 2     private static IoBufferAllocator allocator = new SimpleBufferAllocator();
 3 
 4     /** 是否使用直接内存,默认为false表示使用堆内内存,值为true表示使用直接内存 */
 5     private static boolean useDirectBuffer = false;
 6 
 7     public static IoBuffer allocate(int capacity) {
 8         return allocate(capacity, useDirectBuffer);
 9     }
10 
11     public static IoBuffer allocate(int capacity, boolean useDirectBuffer) {
12         if (capacity < 0) {
13             throw new IllegalArgumentException("capacity: " + capacity);
14         }
15         /** 调用内存分配工具的allocate方法进行内存分配 */
16         return allocator.allocate(capacity, useDirectBuffer);
17     }

分配缓冲区一共有两个参数，分别是缓冲区的大小和是否使用直接内存，最终调用内存分配器的allocate方法进行内存分配，源码如下：

 1 /** 创建IoBuffer对象,分配内存 */
 2     public IoBuffer allocate(int capacity, boolean direct) {
 3         return wrap(allocateNioBuffer(capacity, direct));
 4     }
 5 
 6     /** 创建NIO的ByteBuffer对象*/
 7     public ByteBuffer allocateNioBuffer(int capacity, boolean direct) {
 8         ByteBuffer nioBuffer;
 9         if (direct) {
10             /** 分配直接内存*/
11             nioBuffer = ByteBuffer.allocateDirect(capacity);
12         } else {
13             /** 分配堆内内存*/
14             nioBuffer = ByteBuffer.allocate(capacity);
15         }
16         return nioBuffer;
17     }
18 
19     /** 将ByteBuffer对象包装成IoBuffer对象 */
20     public IoBuffer wrap(ByteBuffer nioBuffer) {
21         return new SimpleBuffer(nioBuffer);
22     }

这里就回到了Java的NIO分配内存的逻辑了，MINA分配缓冲区的逻辑实际底层就是调用了Java NIO的ByteBuffer的分配逻辑，根据直接内存还是堆内内存进行内存分配，最终分配了之后再封装成了IoBuffer对象。

2、IO数据读取

1 @Override
2     protected int read(NioSession session, IoBuffer buf) throws Exception {
3         /** 从session中获取channel*/
4         ByteChannel channel = session.getChannel();
5         /** 调用Java NIO的channel.read(ByteBuffer buffer)方法读取数据 */
6         return channel.read(buf.buf());
7     }

读取数据的逻辑也不复杂，同样是直接调用了Java NIO的channel读取Buffer数据的方式进行读取，具体的实现逻辑可以参考Java NIO的channel读取缓冲区数据的逻辑

3、过滤器链对IO数据进行过滤处理

 1 public void fireMessageReceived(Object message) {
 2         if (message instanceof IoBuffer) {
 3             session.increaseReadBytes(((IoBuffer) message).remaining(), System.currentTimeMillis());
 4         }
 5 
 6         callNextMessageReceived(head, session, message);
 7     }
 8 
 9     public final void increaseReadBytes(long increment, long currentTime) {
10         if (increment <= 0) {
11             return;
12         }
13         //统计已读字节数
14         readBytes += increment;
15         //设置上一次读操作时间
16         lastReadTime = currentTime;
17         //设置空闲为0
18         idleCountForBoth.set(0);
19         idleCountForRead.set(0);
20 
21         if (getService() instanceof AbstractIoService) {
22             ((AbstractIoService) getService()).getStatistics().increaseReadBytes(increment, currentTime);
23         }
24     }
25 
26     /** 通知下一个节点处理接收消息事件 */
27     private void callNextMessageReceived(Entry entry, IoSession session, Object message) {
28         try {
29             IoFilter filter = entry.getFilter();
30             NextFilter nextFilter = entry.getNextFilter();
31             /** 依次执行每一个IoFilter的messageReceived方法 */
32             filter.messageReceived(nextFilter, session, message);
33         } catch (Exception e) {
34             fireExceptionCaught(e);
35         } catch (Error e) {
36             fireExceptionCaught(e);
37             throw e;
38         }
39     }

过滤器链路处理消息时，会从过滤器链路的头节点开始依次执行messageReceived方法，直到执行到最后一个过滤器，过滤器链路头节点和尾节点是固定不变的，头节点实现为HeadFilter，尾节点实现为TailFilter。

当处理读事件时，会从HeadFilter开始处理，然后按自定义的过滤器依次执行，最后执行TailFilter的处理；而处理写事件时，顺序完全相反，会从TailFilter开始到HeadFilter结束。所以读数据时最后会执行TailFilter的messageReceived方法，源码如下：

 1  @Override
 2     public void messageReceived(NextFilter nextFilter, IoSession session, Object message) throws Exception {
 3         AbstractIoSession s = (AbstractIoSession) session;
 4 
 5         if (!(message instanceof IoBuffer)) {
 6             s.increaseReadMessages(System.currentTimeMillis());
 7         } else if (!((IoBuffer) message).hasRemaining()) {
 8             s.increaseReadMessages(System.currentTimeMillis());
 9         }
10 
11         // Update the statistics
12         if (session.getService() instanceof AbstractIoService) {
13             ((AbstractIoService) session.getService()).getStatistics().updateThroughput(System.currentTimeMillis());
14         }
15 
16         try {
17             /** 尾部过滤器将消息交给业务处理器IoHandler处理 */
18             session.getHandler().messageReceived(s, message);
19         } finally {
20             if (s.getConfig().isUseReadOperation()) {
21                 s.offerReadFuture(message);
22             }
23         }
24     }

可以看出TailFilter最重要的一个步骤是需要将发送的IO数据交给业务层处理，通过从IoSession中获取IoHandler对象，并调用IoHandler的messageReceived方法交给业务层处理IO数据

4、缓冲区扩容或降容处理

 1 /** 扩容读缓冲区大小为原先的两倍 */
 2     public final void increaseReadBufferSize() {
 3         /** 扩大两倍*/
 4         int newReadBufferSize = getConfig().getReadBufferSize() << 1;
 5         if (newReadBufferSize <= getConfig().getMaxReadBufferSize()) {
 6             getConfig().setReadBufferSize(newReadBufferSize);
 7         } else {
 8             getConfig().setReadBufferSize(getConfig().getMaxReadBufferSize());
 9         }
10 
11         deferDecreaseReadBuffer = true;
12     }
13 
14     /**
15      * 降容读缓冲区大小为原先的一半
16      */
17     public final void decreaseReadBufferSize() {
18         if (deferDecreaseReadBuffer) {
19             deferDecreaseReadBuffer = false;
20             return;
21         }
22 
23         if (getConfig().getReadBufferSize() > getConfig().getMinReadBufferSize()) {
24             /** 缩小一半*/
25             getConfig().setReadBufferSize(getConfig().getReadBufferSize() >>> 1);
26         }
27 
28         deferDecreaseReadBuffer = true;
29     }

配置的读缓冲区大小和实际的IO数据大小可能会存在偏差，所以需要动态的调整读缓冲区的大小，避免缓冲区内存不足或内存浪费，每次扩容都会扩容到原大小的两倍，降容也会缩小到原先的一半

5、连接关闭或连接异常事件处理

 1 public void fireInputClosed() {
 2         /** 获取过滤器链头节点*/
 3         Entry head = this.head;
 4         callNextInputClosed(head, session);
 5     }
 6 
 7     /** 通知下一个节点处理IO输入关闭事件*/
 8     private void callNextInputClosed(Entry entry, IoSession session) {
 9         try {
10             IoFilter filter = entry.getFilter();
11             /** 获取下一个过滤器*/
12             NextFilter nextFilter = entry.getNextFilter();
13             /** 处理IO输入关闭事件*/
14             filter.inputClosed(nextFilter, session);
15         } catch (Throwable e) {
16             fireExceptionCaught(e);
17         }
18     }
19 
20     public void fireExceptionCaught(Throwable cause) {
21         callNextExceptionCaught(head, session, cause);
22     }
23 
24     private void callNextExceptionCaught(Entry entry, IoSession session, Throwable cause) {
25         /** 唤醒关联的Future*/
26         ConnectFuture future = (ConnectFuture) session.removeAttribute(SESSION_CREATED_FUTURE);
27         if (future == null) {
28             try {
29                 IoFilter filter = entry.getFilter();
30                 NextFilter nextFilter = entry.getNextFilter();
31                 /** 执行过滤的异常处理事件*/
32                 filter.exceptionCaught(nextFilter, session, cause);
33             } catch (Throwable e) {
34                 LOGGER.warn("Unexpected exception from exceptionCaught handler.", e);
35             }
36         } else {
37             /** 关闭session*/
38             if (!session.isClosing()) {
39                 // Call the closeNow method only if needed
40                 session.closeNow();
41             }
42 
43             future.setException(cause);
44         }
45     }

处理逻辑基本上一样，都是先获取过滤器链的头节点，开始处理对应的事件，然后再依次调用下一个节点的过滤器的方法。另外最后一个过滤器TailFilter会将对应的事件会最终交给业务层处理，调用IoHandler的对应方法，让业务层处理事件。

总结MINA服务端的工作机制：

1、创建IoService，每个IoService内部有一个线程池用于处理客户端的连接请求，并且有一个IoProcessor池，默认数量为CPU个数+1，用于处理客户端的IO操作

2、IoService内部有一个Selector用于监听客户端的连接请求，连接成功之后会创建IoSession并交给IoProcessor池处理

3、IoProcessor池根据IoSession的ID进行取模算法选取一个IoProcessor和IoSession进行绑定，一个IoProcessor可以绑定多个IoSession，一个IoSession只可以绑定一个IoProcessor

4、IoProcessor内部也有一个Selector用来监控所有关联的IoSession的IO操作状态，并且有一个线程池用来处理IO操作

5、IoProcessor将处理的IO事件交给IoService绑定的过滤器链，过滤器链上的所有过滤器依次处理IO事件

7、过滤器链的最后一个过滤器为TailFilter,处理完IO事件之后将IO事件交给业务层处理器IoHandler

8、IoHandler处理业务数据，而写数据的话顺序完全相反，才业务层到过滤器链路，通过HeadFilter再由IoProcessor发送出去