Mina编程的两个注意点

• 1. 首先，这是一个nio的框架，仍然是采用reactor模式，知道这一点后，那么编程就没有什么难的。nio的编程，无外乎就是这些套路， 再进一步说，网络编程，也就是这些套路了。
• 2. 那么剩下编程的注意点，也就是编解码的处理以及最后的业务逻辑的处理。

2.1 编解码的注意点：因为在网络编程中，client和server之间，往往需要完整的接收到一条消息的后，才交给业务逻辑处理。具体可以参看http://jimmee./blog/617544，其中，我们常常是继承自CumulativeProtocolDecoder来实现自己的解码器，主要的docode的方法，其作用是将本次数据和上次接收到的数据（如果doDecode方法没有处理完的话），统一放到一个buffer中，之后扔给 doDecode方法处理，若处理完之后，还有剩下的数据，则继续缓存。 

/**
    * Cumulates content of <tt>in</tt> into internal buffer and forwards
    * decoding request to {@link #doDecode(IoSession, IoBuffer, ProtocolDecoderOutput)}.
    * <tt>doDecode()</tt> is invoked repeatedly until it returns <tt>false</tt>
    * and the cumulative buffer is compacted after decoding ends.
    *
    * @throws IllegalStateException if your <tt>doDecode()</tt> returned
    *                               <tt>true</tt> not consuming the cumulative buffer.
    */
   public void decode(IoSession session, IoBuffer in,
           ProtocolDecoderOutput out) throws Exception {
       if (!session.getTransportMetadata().hasFragmentation()) {
           while (in.hasRemaining()) {
               if (!doDecode(session, in, out)) {
                   break;
               }
           }

           return;
       }

       boolean usingSessionBuffer = true;
       IoBuffer buf = (IoBuffer) session.getAttribute(BUFFER);
       // If we have a session buffer, append data to that; otherwise
       // use the buffer read from the network directly.
       if (buf != null) {
           boolean appended = false;
           // Make sure that the buffer is auto-expanded.
           if (buf.isAutoExpand()) {
               try {
                   buf.put(in);
                   appended = true;
               } catch (IllegalStateException e) {
                   // A user called derivation method (e.g. slice()),
                   // which disables auto-expansion of the parent buffer.
               } catch (IndexOutOfBoundsException e) {
                   // A user disabled auto-expansion.
               }
           }

           if (appended) {
               buf.flip();
           } else {
               // Reallocate the buffer if append operation failed due to
               // derivation or disabled auto-expansion.
               buf.flip();
               IoBuffer newBuf = IoBuffer.allocate(
                       buf.remaining() + in.remaining()).setAutoExpand(true);
               newBuf.order(buf.order());
               newBuf.put(buf);
               newBuf.put(in);
               newBuf.flip();
               buf = newBuf;

               // Update the session attribute.
               session.setAttribute(BUFFER, buf);
           }
       } else {
           buf = in;
           usingSessionBuffer = false;
       }

  // 上面操作完后，得到的buf是包含以前积累的数据（如果没有读取处理掉），再加上本次得到
       // 的数据，最后扔给doDecode处理。
       for (;;) {
           int oldPos = buf.position();
           boolean decoded = doDecode(session, buf, out);
           if (decoded) {
               if (buf.position() == oldPos) {
                   throw new IllegalStateException(
                           "doDecode() can't return true when buffer is not consumed.");
               }

               if (!buf.hasRemaining()) {
                   break;
               }
           } else {
               break;
           }
       }

       // if there is any data left that cannot be decoded, we store
       // it in a buffer in the session and next time this decoder is
       // invoked the session buffer gets appended to
       if (buf.hasRemaining()) {
           if (usingSessionBuffer && buf.isAutoExpand()) {
               buf.compact();
           } else {
               storeRemainingInSession(buf, session);
           }
       } else {
           if (usingSessionBuffer) {
               removeSessionBuffer(session);
           }
       }
   }

可以看一个具体的实现：PrefixedStringDecoder的 doDecode方法，这个decode是消息长度+具体字节流的消息格式。 

protected boolean doDecode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception {
       if (in.prefixedDataAvailable(prefixLength, maxDataLength)) {
           String msg = in.getPrefixedString(prefixLength, charset.newDecoder());
           out.write(msg);
           return true;
       }

       return false;
   }

其中， prefixedDataAvailable方法是判断得到IoBuffer里的数据是否满足一条消息了，如果再进去看这个方法的实现化，读取是，使用ByteBuffer的绝对位置的读取方法（这种读取不影响position的值的）；当已经有一条完整的消息时，则用getPrefixedString读取（使用的是ByteBuffer的相对位置的读取方法，这会影响position的值，从而实际的消费掉数据）. 

2.2 业务逻辑的处理注意点，一般都会使用一个线程池处理。这里就有一个问题，有时候同一个连接的消息处理，是希望按照顺序来进行的。这也很简单，不需要保证一个连接的所有的业务处理都限定在一个固定的线程中，但是需要保证当有消息需要处理时，这些消息的处理，都在同一个线程中完成。当然了，mina中也有了相应的实现OrderedThreadPoolExecutor。实现原理很简单： 
一个连接（session）对应一个消息队列，同时此session也放到一个队列中，说明这个session有消息需要处理。具体来说，就是使用SessionTasksQueue来表示一个Session的要处理的消息的队列； 
使用 

/** A queue used to store the available sessions */
  private final BlockingQueue<IoSession> waitingSessions = new LinkedBlockingQueue<IoSession>();

waitingSessions表示有消息需要处理的session的队列。 

/**
     * {@inheritDoc}
     */
    @Override
    public void execute(Runnable task) {
        if (shutdown) {
            rejectTask(task);
        }

        // Check that it's a IoEvent task
        checkTaskType(task);

        IoEvent event = (IoEvent) task;

        // Get the associated session
        IoSession session = event.getSession();

        // 得到保存session消息的队列
        SessionTasksQueue sessionTasksQueue = getSessionTasksQueue(session);
        Queue<Runnable> tasksQueue = sessionTasksQueue.tasksQueue;

        boolean offerSession;

        // propose the new event to the event queue handler. If we
        // use a throttle queue handler, the message may be rejected
        // if the maximum size has been reached.
        boolean offerEvent = eventQueueHandler.accept(this, event);

        if (offerEvent) {
            // Ok, the message has been accepted
            synchronized (tasksQueue) {
                // Inject the event into the executor taskQueue
                tasksQueue.offer(event);

             // 如果session中的消息已经处理完了，说明没有线程在处理这个session，
                 // 重新提交给线程池处理
                if (sessionTasksQueue.processingCompleted) {
                    sessionTasksQueue.processingCompleted = false;
                    offerSession = true;
                } else {
                    offerSession = false;
                }

                if (LOGGER.isDebugEnabled()) {
                    print(tasksQueue, event);
                }
            }
        } else {
            offerSession = false;
        }

        if (offerSession) {
            // As the tasksQueue was empty, the task has been executed
            // immediately, so we can move the session to the queue
            // of sessions waiting for completion.
            waitingSessions.offer(session);
        }

        addWorkerIfNecessary();

        if (offerEvent) {
            eventQueueHandler.offered(this, event);
        }

对应的，看一下： 

private void runTasks(SessionTasksQueue sessionTasksQueue) {
           for (;;) {
               Runnable task;
               Queue<Runnable> tasksQueue = sessionTasksQueue.tasksQueue;

               synchronized (tasksQueue) {
                   task = tasksQueue.poll();
                   // 当已经没有任务时，处理此session的线程结束掉
                   if (task == null) {
                       sessionTasksQueue.processingCompleted = true;
                       break;
                   }
               }

               eventQueueHandler.polled(OrderedThreadPoolExecutor.this, (IoEvent) task);

               runTask(task);
           }
       }

小结：个人认为，只要对nio编程熟悉，对mina框架，只要明白了以上两点，就不再有什么大问题了。