mina2.0 源码分析--- 基于nio的服务端socket监听过程

   整体来说 ，mina服务端采用基于nio的单线程，轮询机制。 使用selector  获取客户端的链接，并创建sesssion，session通过process来处理io操作。

    nio的典型模式如下所示：

 

private NioEchoServer() throws IOException {
       Selector  roller = Selector.open();
        ServerSocketChannel serverChannel = ServerSocketChannel.open();  //
        serverChannel.socket().bind(new InetSocketAddress(port));
        serverChannel.configureBlocking(false);
        serverChannel.register(roller, SelectionKey.OP_ACCEPT);
    }


  public void start() throws IOException {
        int keyAdded = 0;
        while ((keyAdded = roller.select()) > 0) {
            Set<SelectionKey> keySets = roller.selectedKeys();
            Iterator iter = keySets.iterator();
            while (iter.hasNext()) {
                SelectionKey key = (SelectionKey) iter.next();
                iter.remove();
                actionHandler(key);
            }
        }
    }

   1 服务端绑定port

   2 创建Selector

   3 创建SeversocketChannel，面向流的侦听socket 通道。

   4  监听客户端的请求，

 

 

 

接下来就通过mina 创建服务端socket Accpeptor 使用单线程并监处理端口和客户端请求，学习一下mina的源码

IoAcceptor  acceptor  =  new  NioSoketAcceptor ();  //根据实现的类，调用不同的构造方法

 

上面代码， 创建一个socket Acceptor  ，mina内部实现如下 ：

 

super(new DefaultSocketSessionConfig(), NioProcessor.class); //传入具体的ioprocess 的实现类。

 

最终调用的类和构造方法：

AbstractPollingIoAcceptor     
   protected AbstractPollingIoAcceptor(IoSessionConfig sessionConfig,    //模板模式的使用，根据子类传入IoProcessor实现类构造SimpleIoprocessorPool 池
            Class<? extends IoProcessor<T>> processorClass) {
        this(sessionConfig, null, new SimpleIoProcessorPool<T>(processorClass),
                true);
    }

 

  SimpleIoProcessorPool 创建线程池，提供给Iosession ，做具体的io处理工作。

public SimpleIoProcessorPool(Class<? extends IoProcessor<T>> processorType) {

        this(processorType, null, DEFAULT_SIZE);

   }
     public SimpleIoProcessorPool(Class<? extends IoProcessor<T>> processorType, Executor executor, int size) { //构造IOprocessorPool 中的IOprocessor ，并指定Executor；
    try {
          .....................
          .....................   
         //创建一个可根据需要创建新线程的线程池，但是在以前构造的线程可用时将重用它们，并在需要时使用提供的 ThreadFactory 创建新线程
          this.executor = executor = Executors.newCachedThreadPool();   //return  new ThreadPoolExecutor ;
          ...............................
          for (int i = 0; i < pool.length; i ++)
                 .........................
                 ...........................
            //指定Executor .,线程池，提供构造线程。 使用构造函数 NioProcess（Executor execytor ）构造processor ，打开selector 。     
                      processor = processorType.getConstructor(ExecutorService.class).newInstance(executor);
                .............................
                ..................................
                 pool[i] = processor;
                      .....................
                    } catch (NoSuchMethodException e) {
   

 

       
   应用代码中 调用acceptor .bind（），完成对特定端口的绑定，开始监听。
    nioScoket bind （）  调用  AbstarctIoProcessor .bind() ,调用 AbstractPollingIo.bindInternal()

完成对客户端的监听。

public final void bind(Iterable<? extends SocketAddress> localAddresses) throws IOException {
  {
      .......................
            .....................
      try {
                boundAddresses.addAll(bindInternal(localAddressesCopy));  //模板模式
            } catch (IOException e) {
                throw e;
            } catch (RuntimeException e) {
                throw e;
            } catch (Throwable e) {
                throw new RuntimeIoException(
                        "Failed to bind to: " + getLocalAddresses(), e);
            }
        }
                  ...................
        ....................
  }
 
AbstractPollingIoacceptor  ：
  protected final Set<SocketAddress> bindInternal( List<? extends SocketAddress> localAddresses) throws Exception {

        AcceptorOperationFuture request = new AcceptorOperationFuture( localAddresses);

       registerQueue.add(request);


    //   acceptor 是一个接受用户请求的线程。It's a thread accepting incoming connections from clients.
    //  executor.execute(new NamePreservingRunnable(acceptor, actualThreadName));
     //启动接受用户请求。
        startupAcceptor();            

        wakeup();      //selector.wakeup(); 使尚未返回的第一个选择操作立即返回。

        request.awaitUninterruptibly();  ////堵塞直至监听成功 

        if (request.getException() != null) {

             throw request.getException();
        }
        Set<SocketAddress> newLocalAddresses = new HashSet<SocketAddress>();
        
        for (H handle:boundHandles.values()) {
            newLocalAddresses.add(localAddress(handle));   
            return newLocalAddresses;

  }

 

AbstractPollingIoAcceptor 的bindInternal   （。。。） 调用 startupAcceptor，执行 acceptor 线程 。

 

使用自己的私有类 ，private class Acceptor implements Runnable  ，来处理接收客户端请求工作

 

 private void startupAcceptor() {
        // If the acceptor is not ready, clear the queues
        // TODO : they should already be clean : do we have to do that ?
        if (!selectable) {
            registerQueue.clear();
            cancelQueue.clear();
        }

        // start the acceptor if not already started
        synchronized (lock) {
            if (acceptor == null) {
                acceptor = new Acceptor();
                executeWorker(acceptor);
            }
        }
    }

 

 

循环处理客户端请求。

acceptor  ，work线程代码

/**
     * This class is called by the startupAcceptor() method and is
     * placed into a NamePreservingRunnable class.
     * It's a thread accepting incoming connections from clients.
     * The loop is stopped when all the bound handlers are unbound.
     */
    private class Acceptor implements Runnable {
        public void run() {
            int nHandles = 0;

            while (selectable) {
                try {
                    // Detect if we have some keys ready to be processed
                    // The select() will be woke up if some new connection
                    // have occurred, or if the selector has been explicitly
                    // woke up
                    int selected = select();

                    // this actually sets the selector to OP_ACCEPT,
                    // and binds to the port on which this class will
                    // listen on
                    nHandles += registerHandles();

                    if (selected > 0) {
                        // We have some connection request, let's process 
                        // them here. 
                        processHandles(selectedHandles());
                    }

                    // check to see if any cancellation request has been made.
                    nHandles -= unregisterHandles();

                    // Now, if the number of registred handles is 0, we can
                    // quit the loop: we don't have any socket listening
                    // for incoming connection.
                    if (nHandles == 0) {
                        synchronized (lock) {
                            if (registerQueue.isEmpty()
                                    && cancelQueue.isEmpty()) {
                                acceptor = null;
                                break;
                            }
                        }
                    }
                } catch (Throwable e) {
                    ExceptionMonitor.getInstance().exceptionCaught(e);

                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e1) {
                        ExceptionMonitor.getInstance().exceptionCaught(e1);
                    }
                }
            }

            // Cleanup all the processors, and shutdown the acceptor.
            if (selectable && isDisposing()) {
                selectable = false;
                try {
                    if (createdProcessor) {
                        processor.dispose();
                    }
                } finally {
                    try {
                        synchronized (disposalLock) {
                            if (isDisposing()) {
                                destroy();
                            }
                        }
                    } catch (Exception e) {
                        ExceptionMonitor.getInstance().exceptionCaught(e);
                    } finally {
                        disposalFuture.setDone();
                    }
                }
            }
        }