10. 处理OP_READ

  mina为每个session映射到一个NioProcessor，NioProcessor是一个IO处理机，会开启一个线程Processor，也持有一个专门的Selector，将其拥有的Session的OP_READ注册到该Selector上。因为read返回-1表示断开连接，所以这里也需要处理断开事件。
代码跟踪如下：

private class Processor implements Runnable {
     public void run() {
     ...
     for (;;) {
          try {
          ...
         int selected = select(SELECT_TIMEOUT);
          ...
          if (selected > 0) {
              //LOG.debug("Processing ..."); // This log hurts one of the MDCFilter test...
              process();
          }

         ...
         } catch (ClosedSelectorException cse) {
              break;
         } catch (Throwable t) {
              ...
        }
     }
     ...
    }
}
private void process() throws Exception {
        for (Iterator<S> i = selectedSessions(); i.hasNext();) {
            S session = i.next();
            process(session);
            i.remove();
        }
    }

 private void process(S session) {
        // Process Reads
        if (isReadable(session) && !session.isReadSuspended()) {
            read(session);
        }

        // Process writes
        if (isWritable(session) && !session.isWriteSuspended()) {
            // add the session to the queue, if it's not already there
            if (session.setScheduledForFlush(true)) {
                flushingSessions.add(session);
            }
        }
    }

关键代码在于read(session函数中)：

private void read(S session) {
        IoSessionConfig config = session.getConfig();
        int bufferSize = config.getReadBufferSize();
        //申请一个buffer，接受到数据存储在这里
        IoBuffer buf = IoBuffer.allocate(bufferSize);
        //是否处理分包--默认为true
        final boolean hasFragmentation = session.getTransportMetadata().hasFragmentation();

        try {
            int readBytes = 0;
            int ret;
            try {
                if (hasFragmentation) {
                    //如果处理分包，则读到read返回0，或者buffer被填满
                    //这样read一般被读两次
                    while ((ret = read(session, buf)) > 0) {
                        readBytes += ret;
                        if (!buf.hasRemaining()) {
                            break;
                        }
                    }
                } else {
                    //不处理分包，直接读一次就行
                    ret = read(session, buf);
                    if (ret > 0) {
                        readBytes = ret;
                    }
                }
            } finally {
                //buffer写完毕，转为读模式
                buf.flip();
            }
            if (readBytes > 0) {
                //读到的数据需要经过一个过滤链，然后有链尾节点转到handle处理
                IoFilterChain filterChain = session.getFilterChain();
                //触发数据接受事件
                filterChain.fireMessageReceived(buf);
                buf = null;
                //调整接受缓冲区大小，该缓冲区非socket的缓冲区
                if (hasFragmentation) {
                    if (readBytes << 1 < config.getReadBufferSize()) {
                        session.decreaseReadBufferSize();
                    } else if (readBytes == config.getReadBufferSize()) {
                        session.increaseReadBufferSize();
                    }
                }
            }
            //处理连接断开事件
            if (ret < 0) {
                scheduleRemove(session);
            }
        } catch (Throwable e) {
            if (e instanceof IOException) {
                if (!(e instanceof PortUnreachableException)
                        || !AbstractDatagramSessionConfig.class.isAssignableFrom(config.getClass())
                        || ((AbstractDatagramSessionConfig) config).isCloseOnPortUnreachable()) {
                    scheduleRemove(session);
                }
            }
            //抛出异常，则触发抛出异常事件
            IoFilterChain filterChain = session.getFilterChain();
            filterChain.fireExceptionCaught(e);
        }
    }

二. 连接断开

private void scheduleRemove(S session) {
        removingSessions.add(session);
}

op_read读返回-1，只是简单的将seesion加入removingSessions队列。处理逻辑在Processor线程：

private int removeSessions() {
        int removedSessions = 0;

        for (S session = removingSessions.poll(); session != null; session = removingSessions.poll()) {
            SessionState state = getState(session);

            switch (state) {
            case OPENED:
                // Try to remove this session
                if (removeNow(session)) {
                    removedSessions++;
                }

                break;

            case CLOSING:
                // Skip if channel is already closed
                break;

            case OPENING:
                // Remove session from the newSessions queue and
                // remove it
                newSessions.remove(session);

                if (removeNow(session)) {
                    removedSessions++;
                }

                break;

            default:
                throw new IllegalStateException(String.valueOf(state));
            }
        }

        return removedSessions;
    }

private boolean removeNow(S session) {
        clearWriteRequestQueue(session);

        try {
            destroy(session);
            return true;
        } catch (Exception e) {
            IoFilterChain filterChain = session.getFilterChain();
            filterChain.fireExceptionCaught(e);
        } finally {
            clearWriteRequestQueue(session);
            ((AbstractIoService) session.getService()).getListeners().fireSessionDestroyed(session);
        }
        return false;
    }

protected void destroy(NioSession session) throws Exception {
        ByteChannel ch = session.getChannel();
        SelectionKey key = session.getSelectionKey();
        if (key != null) {
            key.cancel();
        }
        ch.close();
    }