//Listener构造函数public Listener() throws IOException {      address = new InetSocketAddress(bindAddress, port);      // Create a new server socket and set to non blocking mode      acceptChannel = ServerSocketChannel.open();      acceptChannel.configureBlocking(false);      // Bind the server socket to the local host and port      bind(acceptChannel.socket(), address, backlogLength);      port = acceptChannel.socket().getLocalPort(); //Could be an ephemeral port      // create a selector;      selector= Selector.open();      readers = new Reader[readThreads];      readPool = Executors.newFixedThreadPool(readThreads);      for (int i = 0; i < readThreads; i++) {        Selector readSelector = Selector.open();        Reader reader = new Reader(readSelector);        readers[i] = reader;        readPool.execute(reader);      }

//Listener运行函数public void run() {      LOG.info(getName() + ": starting");      SERVER.set(Server.this);      while (running) {        SelectionKey key = null;        try {          selector.select();          Iterator<SelectionKey> iter = selector.selectedKeys().iterator();          while (iter.hasNext()) {            key = iter.next();            iter.remove();            try {              if (key.isValid()) {                if (key.isAcceptable())                  doAccept(key);              }            } catch (IOException e) {            }            key = null;          }        } catch (OutOfMemoryError e) {          // we can run out of memory if we have too many threads          // log the event and sleep for a minute and give           // some thread(s) a chance to finish          LOG.warn("Out of Memory in server select", e);          closeCurrentConnection(key, e);          cleanupConnections(true);          try { Thread.sleep(60000); } catch (Exception ie) {}        } catch (Exception e) {          closeCurrentConnection(key, e);        }        cleanupConnections(false);      }      LOG.info("Stopping " + this.getName());      synchronized (this) {        try {          acceptChannel.close();          selector.close();        } catch (IOException e) { }        selector= null;        acceptChannel= null;                // clean up all connections        while (!connectionList.isEmpty()) {          closeConnection(connectionList.remove(0));        }      }    }

//Do_Acceptvoid doAccept(SelectionKey key) throws IOException,  OutOfMemoryError {      Connection c = null;      ServerSocketChannel server = (ServerSocketChannel) key.channel();      SocketChannel channel;      while ((channel = server.accept()) != null) {        channel.configureBlocking(false);        channel.socket().setTcpNoDelay(tcpNoDelay);        Reader reader = getReader();        try {          reader.startAdd();          SelectionKey readKey = reader.registerChannel(channel);          c = new Connection(readKey, channel, System.currentTimeMillis());          readKey.attach(c);          synchronized (connectionList) {            connectionList.add(numConnections, c);            numConnections++;          }          if (LOG.isDebugEnabled())            LOG.debug("Server connection from " + c.toString() +                "; # active connections: " + numConnections +                "; # queued calls: " + callQueue.size());                  } finally {          reader.finishAdd();         }      }    }

public synchronized SelectionKey registerChannel(SocketChannel channel)                                                          throws IOException {          return channel.register(readSelector, SelectionKey.OP_READ);      }