Io不再神秘

ServerSocket server = new ServerSocket(port);
  while(true) {
    Socket connection = server.accept();
    spawn-Thread-and-process(connection);
  }

elector selector = Selector.open();
   
  channel.configureBlocking(false);
   
  SelectionKey key = channel.register(selector, SelectionKey.OP_READ);
   
  while(true) {
   
    int readyChannels = selector.select();
   
    if(readyChannels == 0) continue;
   
    Set<SelectionKey> selectedKeys = selector.selectedKeys();
   
    Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
   
    while(keyIterator.hasNext()) {
   
      SelectionKey key = keyIterator.next();
   
      if(key.isAcceptable()) {
          // a connection was accepted by a ServerSocketChannel.
   
      } else if (key.isConnectable()) {
          // a connection was established with a remote server.
   
      } else if (key.isReadable()) {
          // a channel is ready for reading
   
      } else if (key.isWritable()) {
          // a channel is ready for writing
      }
   
      keyIterator.remove();
    }
  }

public class ReactorInitiator {
   
  private static final int NIO_SERVER_PORT = 9993;
   
    public void initiateReactiveServer(int port) throws Exception {
   
      ServerSocketChannel server = ServerSocketChannel.open();
      server.socket().bind(new InetSocketAddress(port));
      server.configureBlocking(false);
   
      Dispatcher dispatcher = new Dispatcher();
      dispatcher.registerChannel(SelectionKey.OP_ACCEPT, server);
   
      dispatcher.registerEventHandler(
        SelectionKey.OP_ACCEPT, new AcceptEventHandler(
        dispatcher.getDemultiplexer()));
   
      dispatcher.registerEventHandler(
        SelectionKey.OP_READ, new ReadEventHandler(
        dispatcher.getDemultiplexer()));
   
      dispatcher.registerEventHandler(
      SelectionKey.OP_WRITE, new WriteEventHandler());
   
      dispatcher.run(); // Run the dispatcher loop
   
   }
   
    public static void main(String[] args) throws Exception {
      System.out.println("Starting NIO server at port : " +
        NIO_SERVER_PORT);
      new ReactorInitiator().
        initiateReactiveServer(NIO_SERVER_PORT);
    }
   
  }
   
  public class Dispatcher {
   
    private Map<Integer, EventHandler> registeredHandlers =
      new ConcurrentHashMap<Integer, EventHandler>();
    private Selector demultiplexer;
   
    public Dispatcher() throws Exception {
      demultiplexer = Selector.open();
    }
   
    public Selector getDemultiplexer() {
      return demultiplexer;
    }
   
    public void registerEventHandler(
      int eventType, EventHandler eventHandler) {
      registeredHandlers.put(eventType, eventHandler);
    }
   
    // Used to register ServerSocketChannel with the
    // selector to accept incoming client connections
    public void registerChannel(
      int eventType, SelectableChannel channel) throws Exception {
      channel.register(demultiplexer, eventType);
    }
   
    public void run() {
      try {
        while (true) { // Loop indefinitely
          demultiplexer.select();
   
          Set<SelectionKey> readyHandles =
            demultiplexer.selectedKeys();
          Iterator<SelectionKey> handleIterator =
            readyHandles.iterator();
   
          while (handleIterator.hasNext()) {
            SelectionKey handle = handleIterator.next();
   
            if (handle.isAcceptable()) {
              EventHandler handler =
                registeredHandlers.get(SelectionKey.OP_ACCEPT);
                handler.handleEvent(handle);
             // Note : Here we don't remove this handle from
             // selector since we want to keep listening to
             // new client connections
            }
   
            if (handle.isReadable()) {
              EventHandler handler =
                registeredHandlers.get(SelectionKey.OP_READ);
              handler.handleEvent(handle);
              handleIterator.remove();
            }
   
            if (handle.isWritable()) {
              EventHandler handler =
                registeredHandlers.get(SelectionKey.OP_WRITE);
              handler.handleEvent(handle);
              handleIterator.remove();
            }
          }
        }
      } catch (Exception e) {
        e.printStackTrace();
      }
    }
   
  }
   
  public interface EventHandler {
   
     public void handleEvent(SelectionKey handle) throws Exception;
   
  }
   
  public class AcceptEventHandler implements EventHandler {
    private Selector demultiplexer;
    public AcceptEventHandler(Selector demultiplexer) {
      this.demultiplexer = demultiplexer;
    }
   
    @Override
    public void handleEvent(SelectionKey handle) throws Exception {
      ServerSocketChannel serverSocketChannel =
       (ServerSocketChannel) handle.channel();
      SocketChannel socketChannel = serverSocketChannel.accept();
      if (socketChannel != null) {
        socketChannel.configureBlocking(false);
        socketChannel.register(
          demultiplexer, SelectionKey.OP_READ);
      }
    }
   
  }
   
  public class ReadEventHandler implements EventHandler {
   
    private Selector demultiplexer;
    private ByteBuffer inputBuffer = ByteBuffer.allocate(2048);
   
    public ReadEventHandler(Selector demultiplexer) {
      this.demultiplexer = demultiplexer;
    }
   
    @Override
    public void handleEvent(SelectionKey handle) throws Exception {
      SocketChannel socketChannel =
       (SocketChannel) handle.channel();
   
      socketChannel.read(inputBuffer); // Read data from client
   
      inputBuffer.flip();
      // Rewind the buffer to start reading from the beginning
   
      byte[] buffer = new byte[inputBuffer.limit()];
      inputBuffer.get(buffer);
   
      System.out.println("Received message from client : " +
        new String(buffer));
      inputBuffer.flip();
      // Rewind the buffer to start reading from the beginning
      // Register the interest for writable readiness event for
      // this channel in order to echo back the message
   
      socketChannel.register(
        demultiplexer, SelectionKey.OP_WRITE, inputBuffer);
    }
   
  }
   
  public class WriteEventHandler implements EventHandler {
   
    @Override
    public void handleEvent(SelectionKey handle) throws Exception {
      SocketChannel socketChannel =
        (SocketChannel) handle.channel();
      ByteBuffer inputBuffer = (ByteBuffer) handle.attachment();
      socketChannel.write(inputBuffer);
      socketChannel.close(); // Close connection
    } 
  }

public class ProactorInitiator {
    static int ASYNC_SERVER_PORT = 4333;
   
    public void initiateProactiveServer(int port)
      throws IOException {
   
      final AsynchronousServerSocketChannel listener =
        AsynchronousServerSocketChannel.open().bind(
          new InetSocketAddress(port));
       AcceptCompletionHandler acceptCompletionHandler =
         new AcceptCompletionHandler(listener);
   
       SessionState state = new SessionState();
       listener.accept(state, acceptCompletionHandler);
    }
   
    public static void main(String[] args) {
      try {
         System.out.println("Async server listening on port : " +
           ASYNC_SERVER_PORT);
         new ProactorInitiator().initiateProactiveServer(
           ASYNC_SERVER_PORT);
      } catch (IOException e) {
       e.printStackTrace();
      }
   
      // Sleep indefinitely since otherwise the JVM would terminate
      while (true) {
        try {
          Thread.sleep(Long.MAX_VALUE);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    }
  }
   
  public class AcceptCompletionHandler
    implements
      CompletionHandler<AsynchronousSocketChannel, SessionState> {
   
    private AsynchronousServerSocketChannel listener;
   
    public AcceptCompletionHandler(
      AsynchronousServerSocketChannel listener) {
      this.listener = listener;
    }
   
    @Override
    public void completed(AsynchronousSocketChannel socketChannel,
      SessionState sessionState) {
     // accept the next connection
     SessionState newSessionState = new SessionState();
     listener.accept(newSessionState, this);
   
     // handle this connection
     ByteBuffer inputBuffer = ByteBuffer.allocate(2048);
     ReadCompletionHandler readCompletionHandler =
       new ReadCompletionHandler(socketChannel, inputBuffer);
     socketChannel.read(
       inputBuffer, sessionState, readCompletionHandler);
    }
   
    @Override
    public void failed(Throwable exc, SessionState sessionState) {
     // Handle connection failure...
    }
   
  }
   
  public class ReadCompletionHandler implements
    CompletionHandler<Integer, SessionState> {
   
     private AsynchronousSocketChannel socketChannel;
     private ByteBuffer inputBuffer;
   
     public ReadCompletionHandler(
       AsynchronousSocketChannel socketChannel,
       ByteBuffer inputBuffer) {
       this.socketChannel = socketChannel;
       this.inputBuffer = inputBuffer;
     }
   
     @Override
     public void completed(
       Integer bytesRead, SessionState sessionState) {
   
       byte[] buffer = new byte[bytesRead];
       inputBuffer.rewind();
       // Rewind the input buffer to read from the beginning
   
       inputBuffer.get(buffer);
       String message = new String(buffer);
   
       System.out.println("Received message from client : " +
         message);
   
       // Echo the message back to client
       WriteCompletionHandler writeCompletionHandler =
         new WriteCompletionHandler(socketChannel);
   
       ByteBuffer outputBuffer = ByteBuffer.wrap(buffer);
   
       socketChannel.write(
         outputBuffer, sessionState, writeCompletionHandler);
    }
   
    @Override
    public void failed(Throwable exc, SessionState attachment) {
      //Handle read failure.....
     }
   
  }
   
  public class WriteCompletionHandler implements
    CompletionHandler<Integer, SessionState> {
   
    private AsynchronousSocketChannel socketChannel;
   
    public WriteCompletionHandler(
      AsynchronousSocketChannel socketChannel) {
      this.socketChannel = socketChannel;
    }
   
    @Override
    public void completed(
      Integer bytesWritten, SessionState attachment) {
      try {
        socketChannel.close();
      } catch (IOException e) {
        e.printStackTrace();
      }
    }
   
    @Override
    public void failed(Throwable exc, SessionState attachment) {
     // Handle write failure.....
    }
   
  }
   
  public class SessionState {
   
    private Map<String, String> sessionProps =
      new ConcurrentHashMap<String, String>();
   
     public String getProperty(String key) {
       return sessionProps.get(key);
     }
   
     public void setProperty(String key, String value) {
       sessionProps.put(key, value);
     }
   
  }

public class NHttpServer {
   
    public void start() throws IOReactorException {
      HttpParams params = new BasicHttpParams();
      // Connection parameters
      params.
        setIntParameter(
          HttpConnectionParams.SO_TIMEOUT, 60000)
       .setIntParameter(
         HttpConnectionParams.SOCKET_BUFFER_SIZE, 8 * 1024)
       .setBooleanParameter(
         HttpConnectionParams.STALE_CONNECTION_CHECK, true)
       .setBooleanParameter(
         HttpConnectionParams.TCP_NODELAY, true);
   
      final DefaultListeningIOReactor ioReactor =
        new DefaultListeningIOReactor(2, params);
      // Spawns an IOReactor having two reactor threads
      // running selectors. Number of threads here is
      // usually matched to the number of processor cores
      // in the system
   
      // Application specific readiness event handler
      ServerHandler handler = new ServerHandler();
   
      final IOEventDispatch ioEventDispatch =
        new DefaultServerIOEventDispatch(handler, params);
      // Default IO event dispatcher encapsulating the
      // event handler
   
      ListenerEndpoint endpoint = ioReactor.listen(
        new InetSocketAddress(4444));
   
      // start the IO reactor in a new separate thread
      Thread t = new Thread(new Runnable() {
        public void run() {
          try {
            System.out.println("Listening in port 4444");
            ioReactor.execute(ioEventDispatch);
          } catch (InterruptedIOException ex) {
            ex.printStackTrace();
          } catch (IOException e) {
            e.printStackTrace();
          } catch (Exception e) {
            e.printStackTrace();
          }
      }
      });
      t.start();
   
      // Wait for the endpoint to become ready,
      // i.e. for the listener to start accepting requests.
      try {
        endpoint.waitFor();
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
   
    public static void main(String[] args)
      throws IOReactorException {
      new NHttpServer().start();
    }
   
  }
   
  public class ServerHandler implements NHttpServiceHandler {
   
   private static final int BUFFER_SIZE = 2048;
   
   private static final String RESPONSE_SOURCE_BUFFER =
   "response-source-buffer";
   
   // the factory to create HTTP responses
   private final HttpResponseFactory responseFactory;
   
   // the HTTP response processor
   private final HttpProcessor httpProcessor;
   
   // the strategy to re-use connections
   private final ConnectionReuseStrategy connStrategy;
   
   // the buffer allocator
   private final ByteBufferAllocator allocator;
   
   public ServerHandler() {
     super();
     this.responseFactory = new DefaultHttpResponseFactory();
     this.httpProcessor = new BasicHttpProcessor();
     this.connStrategy = new DefaultConnectionReuseStrategy();
     this.allocator = new HeapByteBufferAllocator();
   }
   
   @Override
   public void connected(
     NHttpServerConnection nHttpServerConnection) {
     System.out.println("New incoming connection");
   }
   
   @Override
   public void requestReceived(
     NHttpServerConnection nHttpServerConnection) {
   
     HttpRequest request =
       nHttpServerConnection.getHttpRequest();
     if (request instanceof HttpEntityEnclosingRequest) {
       // Handle POST and PUT requests
     } else {
   
       ContentOutputBuffer outputBuffer =
         new SharedOutputBuffer(
           BUFFER_SIZE, nHttpServerConnection, allocator);
   
       HttpContext context =
         nHttpServerConnection.getContext();
       context.setAttribute(
         RESPONSE_SOURCE_BUFFER, outputBuffer);
       OutputStream os =
         new ContentOutputStream(outputBuffer);
   
       // create the default response to this request
       ProtocolVersion httpVersion =
       request.getRequestLine().getProtocolVersion();
       HttpResponse response =
         responseFactory.newHttpResponse(
           httpVersion, HttpStatus.SC_OK,
           nHttpServerConnection.getContext());
   
       // create a basic HttpEntity using the source
       // channel of the response pipe
       BasicHttpEntity entity = new BasicHttpEntity();
       if (httpVersion.greaterEquals(HttpVersion.HTTP_1_1)) {
         entity.setChunked(true);
       }
       response.setEntity(entity);
   
       String method = request.getRequestLine().
         getMethod().toUpperCase();
   
       if (method.equals("GET")) {
         try {
           nHttpServerConnection.suspendInput();
           nHttpServerConnection.submitResponse(response);
           os.write(new String("Hello client..").
             getBytes("UTF-8"));
   
           os.flush();
           os.close();
       } catch (Exception e) {
         e.printStackTrace();
       }
      } // Handle other http methods
     }
   }
   
   @Override
   public void inputReady(
      NHttpServerConnection nHttpServerConnection,
      ContentDecoder contentDecoder) {
      // Handle request enclosed entities here by reading
      // them from the channel
   }
   
   @Override
   public void responseReady(
      NHttpServerConnection nHttpServerConnection) {
   
     try {
       nHttpServerConnection.close();
     } catch (IOException e) {
       e.printStackTrace();
     }
   }
   
   @Override
   public void outputReady(
     NHttpServerConnection nHttpServerConnection,
     ContentEncoder encoder) {
     HttpContext context = nHttpServerConnection.getContext();
     ContentOutputBuffer outBuf =
      (ContentOutputBuffer) context.getAttribute(
        RESPONSE_SOURCE_BUFFER);
   
     try {
       outBuf.produceContent(encoder);
     } catch (IOException e) {
       e.printStackTrace();
     }
   }
   
   @Override
   public void exception(
     NHttpServerConnection nHttpServerConnection,
     IOException e) {
     e.printStackTrace();
   }
   
   @Override
   public void exception(
     NHttpServerConnection nHttpServerConnection,
     HttpException e) {
     e.printStackTrace();
   }
   
   @Override
   public void timeout(
     NHttpServerConnection nHttpServerConnection) {
     try {
       nHttpServerConnection.close();
     } catch (IOException e) {
       e.printStackTrace();
     }
   }
   
   @Override
   public void closed(
     NHttpServerConnection nHttpServerConnection) {
     try {
       nHttpServerConnection.close();
     } catch (IOException e) {
       e.printStackTrace();
     }
   }
   
  }