非阻塞NIO总结及对比网络echo示例

非阻塞NIO使用场景：

　　NIO是Java提供的非阻塞I/O API。

　　非阻塞的意义在于可以使用一个线程对大量的数据连接进行处理,非常适用于"短数据长连接"的应用场景,例如即时通讯软件。

　　在一个阻塞C/S系统中,服务器要为每一个客户连接开启一个线程阻塞等待客户端发送的消息.若使用非阻塞技术,服务器可以使用一个线程对连接进行轮询,无须阻塞等待.这大大减少了内存资源的浪费,也避免了服务器在客户线程中不断切换带来的CPU消耗,服务器对CPU的有效使用率大大提高.

　　与原始IO对比，异步IO是一种没有阻塞地读写数据的方法，通常，在原始IO中，在代码进行read()调用时，代码会阻塞直至有可供读取的数据。同样，writer()调用将会阻塞直至数据能够写入。而异步 I/O 调用不会阻塞。你可以注册对特定 I/O 事件的监听 ― 可读的数据的到达、新的套接字连接等等，而在发生这样的事件时，系统将会告诉你。异步 I/O 的一个优势在于，它允许你同时根据大量的输入和输出执行 I/O。同步程序常常需要轮询，或者创建许许多多的线程以处理大量的连接。使用异步 I/O，你可以监听任何数量的通道上的事件，不用轮询，也不用额外的线程。

NIO 工具包的成员

Buffer（缓冲器）

Buffer 类是一个抽象类，它有7 个子类分别对应于七种基本的数据类型：ByteBuffer、CharBuffer、DoubleBuffer、FloatBuffer、IntBuffer、LongBuffer 和ShortBuffer。每一个Buffer对象相当于一个数据容器，可以把它看作内存中的一个大的数组，用来存储和提取所有基本类型(boolean 型除外)的数据。Buffer 类的核心是一块内存区，可以直接对其执行与内存有关的操作，利用操作系统特性和能力提高和改善Java 传统I/O 的性能。

Channel（通道）

Channel 被认为是NIO 工具包的一大创新点，是(Buffer)缓冲器和I/O 服务之间的通道，具有双向性，既可以读入也可以写出，可以更高效的传递数据。主要讨论ServerSocketChannel 和SocketChannel，它们都继承了SelectableChannel，是可选择的通道，分别可以工作在同步和异步两种方式下（这里的可选择不是指可以选择两种工作方式，而是指可以有选择的注册自己感兴趣的事件）。当通道工作在同步方式时，它的功能和编程方法与传统的ServerSocket、Socket 对象相似；当通道工作在异步工作方式时，进行输入输出处理不必等到输入输出完毕才返回，并且可以将其感兴趣的（如：接受操作、连接操作、读出操作、写入操作）事件注册到Selector 对象上，与Selector 对象协同工作可以更有效率的支持和管理并发的网络套接字连接。

Channel是I/O通道,可以向其注册Selector,应用成功可以通过select操作获取当前通道已经准备好的可以无阻塞执行的操作.这由SelectionKey表示。

通道既可提取放在缓冲区中的数据（写），也可向缓冲区存入数据供读取（读）。此外，还有一种特殊类型的缓冲区，用于内存映射文件。

Channel 的状态：

可连（ Connectable ）：当一个 Channel 完成 socket 连接操作已完成或者已失败放弃时。

能连（ Acceptable ）：当一个 Channel 已经准备好接受一个新的 socket 连接时。

可读（ Readable ）：当一个 Channel 能被读时。

可写（ Writable ）：当一个 Channel 能被写时。

 各类 Buffer 是数据的容器对象；各类Channel 实现在各类Buffer 与各类I/O 服务间传输数据。Selector 是实现并发型非阻塞I/O 的核心，各种可选择的通道将其感兴趣的事件注册到Selector 对象上，Selector 在一个循环中不断轮循监视这各些注册在其上的Socket 通道。SelectionKey 类则封装了SelectableChannel 对象在Selector 中的注册信息。当Selector 监测到在某个注册的SelectableChannel 上发生了感兴趣的事件时,自动激活产生一个SelectionKey对象,在这个对象中记录了哪一个SelectableChannel 上发生了哪种事件，通过对被激活的SelectionKey 的分析,外界可以知道每个SelectableChannel 发生的具体事件类型,进行相应的处理。

SelectionKey的常量字段SelectionKey.OP_***分别对应Channel的几种操作例如connect(),accept(),read(),write()。select操作后得到SelectionKey.OP_WRITE或者READ即可在Channel上面无阻塞调用read和write方法,Channel的读写操作均需要通过Buffer进行.即读是讲数据从通道中读入Buffer然后做进一步处理.写需要先将数据写入Buffer然后通道接收Buffer。

NIO 有一个主要的类Selector,这个类似一个观察者，只要我们把需要探知的socketchannel告诉Selector,我们接着做别的事情，当有 事件发生时，他会通知我们，传回一组SelectionKey,我们读取这些Key,就会获得我们刚刚注册过的socketchannel,然后，我们从 这个Channel中读取数据，放心，包准能够读到，接着我们可以处理这些数据

NIO 工作原理

　　在并发型服务器程序中使用NIO，实际上是通过网络事件驱动模型实现的。我们应用Select 机制，不用为每一个客户端连接新启线程处理，而是将其注册到特定的Selector 对象上，这就可以在单线程中利用Selector 对象管理大量并发的网络连接，更好的利用了系统资源；采用非阻塞I/O 的通信方式，不要求阻塞等待I/O 操作完成即可返回，从而减少了管理I/O 连接导致的系统开销，大幅度提高了系统性能。

　　当有读或写等任何注册的事件发生时，可以从Selector 中获得相应的SelectionKey ， 从SelectionKey 中可以找到发生的事件和该事件所发生的具体的SelectableChannel，以获得客户端发送过来的数据。由于在非阻塞网络I/O 中采用了事件触发机制，处理程序可以得到系统的主动通知，从而可以实现底层网络I/O 无阻塞、流畅地读写，而不像在原来的阻塞模式下处理程序需要不断循环等待。使用NIO，可以编写出性能更好、更易扩展的并发型服务器程序。

异步IO编程过程　

1.异步 I/O 中的核心对象名为 Selector。Selector 就是你注册对各种 I/O 事件的监听的地方，而且当那些事件发生时，就是这个对象告诉你所发生的事件。所以，我们需要做的第一件事就是创建一个 Selector：Selector selector = Selector.open();然后，我们将对不同的通道对象调用 register() 方法，以便注册我们对这些对象中发生的 I/O 事件的监听。register() 的第一个参数总是这个 Selector。

2.打开一个ServerSocketChannel: 为了接收连接，我们需要一个 ServerSocketChannel。事实上，我们要监听的每一个端口都需要有一个ServerSocketChannel 。对于每一个端口，我们打开一个ServerSocketChannel，如下所示：ServerSocketChannel ssc = ServerSocketChannel.open();
ssc.configureBlocking( false );
ServerSocket ss = ssc.socket();

InetSocketAddress address = new InetSocketAddress( ports[i] );

ss.bind( address );

3.下一步是将新打开的 ServerSocketChannels 注册到 Selector 上。为此我们使用ServerSocketChannel.register() 方法，如下所示：

SelectionKey key = ssc.register( selector, SelectionKey.OP_ACCEPT );register() 的第一个参数总是这个 Selector。第二个参数是 OP_ACCEPT，这里它指定我们想要监听 accept 事件，也就是在新的连接建立时所发生的事件。这是适用于 ServerSocketChannel 的唯一事件类型。
请注意对 register() 的调用的返回值。 SelectionKey 代表这个通道在此 Selector 上的这个注册。当某个Selector 通知你某个传入事件时，它是通过提供对应于该事件的 SelectionKey 来进行的。SelectionKey 还可以用于取消通道的注册。

4.现在已经注册了我们对一些 I/O 事件的兴趣，下面将进入主循环。使用 Selectors 的几乎每个程序都像下面这样使用内部循环：
int num = selector.select();
Set selectedKeys = selector.selectedKeys();
Iterator it = selectedKeys.iterator();
while (it.hasNext()) {
SelectionKey key = (SelectionKey)it.next();
// ... deal with I/O event ...
}
首先，我们调用 Selector 的 select() 方法。这个方法会阻塞，直到至少有一个已注册的事件发生。当一个或者更多的事件发生时， select() 方法将返回所发生的事件的数量。接下来，我们调用 Selector 的 selectedKeys() 方法，它返回发生了事件的 SelectionKey 对象的一个集合 。我们通过迭代 SelectionKeys 并依次处理每个 SelectionKey 来处理事件。对于每一个 SelectionKey，你必须确定发生的是什么 I/O 事件，以及这个事件影响哪些 I/O 对象。

5.监听新连接：

程序执行到这里，我们仅注册了 ServerSocketChannel，并且仅注册它们“接收”事件。为确认这一点，我们对SelectionKey 调用 readyOps() 方法，并检查发生了什么类型的事件：

if (key.isAcceptable())
 {
// Accept the new connection
// ...
}
可以肯定地说， key.isAcceptable() 方法告诉我们该事件是新的连接.

因为我们知道这个服务器套接字上有一个传入连接在等待，所以可以安全地接受它；也就是说，不用担心 accept() 操作会阻塞：
ServerSocketChannel ssc = (ServerSocketChannel)key.channel();
SocketChannel sc = ssc.accept();
下一步是将新连接的 SocketChannel 配置为非阻塞的。而且由于接受这个连接的目的是为了读取来自套接字的数据，所以我们还必须将 SocketChannel 注册到 Selector 上，如下所示：
sc.configureBlocking( false );
SelectionKey newKey = sc.register( selector, SelectionKey.OP_READ );注意我们使用 register() 的 OP_READ 参数，将 SocketChannel 注册用于 读取而不是 接受新连接。

删除处理过的SelectionKey
在处理 SelectionKey 之后，我们几乎可以返回主循环了。但是我们必须首先将处理过的 SelectionKey 从选定的键集合中删除。如果我们没有删除处理过的键，那么它仍然会在主集合中以一个激活的键出现，这会导致我们尝试再次处理它。我们调用迭代器的 remove() 方法来删除处理过的 SelectionKey： it.remove();现在我们可以返回主循环并接受从一个套接字中传入的数据(或者一个传入的 I/O 事件)了。

处理好事件的操作后其余的就是处理IO了。

------------------------------------------------------------------------------------------------------------------------

下面以一个例子演示使用IO和非阻塞NIO来实现这样的操作：从客户端控制台输入一个字符，传到服务器后，服务器给这个字串加一个前缀：（From Server）再传回到客户端

下面是使用普通的IO操作：

服务器端：

public class Server implements Runnable{
    private Socket clientSocket;
    /**
     * @param args
     */
    
    public Server(Socket socket){
        this.clientSocket = socket;
    }
    
    public static void main(String[] args) throws Exception{
        ServerSocket ss = new ServerSocket(8888);
        System.out.println("服务器已经启动，等待连接。。。");
        //通过使用循环，每当有一个客户端连接到此服务器，这服务器启动 一个线程处理与这个客户端的交流
        while(true){
            Socket s = ss.accept();
            System.out.println("一个新的客户端连接建立:"+s.getRemoteSocketAddress());
            new Thread(new Server(s)).start();
        }
    }

    @Override
    public void run() {
        
        try {
            //获得客户端的输入流
            BufferedReader clientReader = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
            //获得到客户端的输出流
            PrintStream clientOut = new PrintStream(clientSocket.getOutputStream());
            boolean flag = true;
            while(flag){
                //读取客户端的数据
                String clientStr = clientReader.readLine();
                //如果从客户端读到的数据为“bye”则停止从这个客户端读取数据
                if("exit".equals(clientStr)){
                    flag = false;
                    break;
                }
                //包装后输出到客户端
                clientOut.println("From Server:"+clientStr);
                
            }
            clientOut.close();
            clientReader.close();
            
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}

客户端：

public class Client {

    /**
     * @param args
     */
    public static void main(String[] args) throws Exception{
        Socket s = new Socket("localhost",8888);
        
        BufferedReader br = new BufferedReader(new InputStreamReader(s.getInputStream()));
        PrintStream ps = new PrintStream(s.getOutputStream());
        
        BufferedReader myInput = new BufferedReader(new InputStreamReader(System.in));
        
        boolean flag = true;
        while(flag){
            System.out.println("输入<exit>退出：");
            //获得要输入到server的数据
            String str = myInput.readLine();
            if("exit".equals(str)){
                flag = false;
                break;
            }
            //把数据输入到server
            ps.println(str);
            //从服务器端获得数据
            System.out.print("来自服务器的数据：");
            String getStr = br.readLine();
            System.out.println(getStr);
            
        }
        myInput.close();
        ps.close();
        br.close();
    }

}

下面是使用非阻塞NIO的操作：

服务器端：

public class NioServer {
    private Selector selector;
    private ByteBuffer buffer;

    public NioServer() throws IOException {
        selector = Selector.open();
        buffer = ByteBuffer.allocate(128);
    }

    public static void main(String[] args) throws Exception {
        new NioServer().buildMonitorInPort("localhost",9999);
    }

    public void buildMonitorInPort(String host,int port) throws Exception {
        InetSocketAddress address = new InetSocketAddress(host,port);
        // 打开ServerSocketChannel通道
        ServerSocketChannel ssc = ServerSocketChannel.open();
        // 配置为非阻塞通道
        ssc.configureBlocking(false);
        // 获得ServerSocket连接
        ServerSocket serverSocket = ssc.socket();
        // 绑定端口
        serverSocket.bind(address);
        // 为此通道注册客户端连接事件
        ssc.register(selector, SelectionKey.OP_ACCEPT);

        // 此循环表示不断地轮询是否有事件发生
        while (true) {
            // 阻塞方法，等待在它上面注册的事件的发生
            int num = selector.select();

            Set<SelectionKey> keys = selector.selectedKeys();
            Iterator<SelectionKey> iter = keys.iterator();

            while (iter.hasNext()) {
                SelectionKey key = iter.next();
                
                if (!key.isValid()) {
                    continue;
                }
                if (key.isAcceptable()) {
                    accept(key);
                }else if (key.isReadable()) {
                    readAndWrite(key);
                }
                // 移除掉key，否则下次还会处理这个key，就会出错
                iter.remove();

            }

        }

    }

    public void accept(SelectionKey key) throws Exception {
        ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
        SocketChannel sc = ssc.accept();
        sc.configureBlocking(false);
        // 注册客户端读的事件
        sc.register(selector, SelectionKey.OP_READ);
        System.out.println("新的连接：（" + sc + "）已建立");
    }

    public void readAndWrite(SelectionKey key) throws Exception {
        // 获得客户端的连接通道
        SocketChannel sc = (SocketChannel) key.channel();
        System.out.println("来自于地址："+sc.socket().getRemoteSocketAddress()+"的信息如下：");
        buffer.clear();
        sc.read(buffer);
        buffer.flip();
        byte[] bytes = new byte[buffer.limit()];
        buffer.get(bytes);
        String s = new String(bytes);
        if(s.equals("exit")){
            System.out.println("来自："+sc.socket().getRemoteSocketAddress()+"断开连接");
            //断开连接之后，要取消键的通道到其选择器的注册
            key.cancel();
            sc.close();
        }else {
            System.out.println(s);
            buffer.clear();
        
            s = "From Server:"+s;
            buffer.put(s.getBytes());
            buffer.flip();
            sc.write(buffer);
        }
    }

}

客户端：

public class NioClient {
    private Selector selector;
    private ByteBuffer buffer;
    private SocketChannel sc;
    //控制客户端的轮询
    private boolean polling = true;
    //用于切换客户端的读写，如果不设置此值，则客户端对于服务端的读写不确定发生，现在是写完之后就读反馈的信息。
    private boolean switchOver = false;
    /**
     * @param args
     */
    public NioClient() throws Exception {
        //初始化selector与buffer
        selector = Selector.open();
        buffer = ByteBuffer.allocate(128);
    }

    public static void main(String[] args) throws Exception {
        new NioClient().buildConnectionInPort("localhost", 9999);
    }

    public void buildConnectionInPort(String address, int port)throws Exception {
        InetSocketAddress address2 = new InetSocketAddress(address, port);
        //打开socketChannel通道并设置为无阻塞通道
        sc = SocketChannel.open();
        sc.configureBlocking(false);
        //建立通道连接
        sc.connect(address2);
        // 注册一个连接事件
        sc.register(selector, SelectionKey.OP_CONNECT);
        //开始轮询连接事件
        while (polling) {
            //阻塞方法，是否有注册事件的发生
            int num = selector.select();
            if(num>0){
                System.out.println("有<"+num+">个事件发生");
                }
            //提取注册事件
            Set<SelectionKey> keys = selector.selectedKeys();
            
            //对注册事件进行处理
            Iterator<SelectionKey> iter = keys.iterator();
            while (iter.hasNext()) {
                SelectionKey key = iter.next();
                System.out.println(key.attachment());
                if (!key.isValid()) {
                    continue;
                }
                //连接事件处理
                if (key.isConnectable()) {
                    connection(key);
                } else if (key.isWritable()||key.isReadable()) {//读写事件处理
                    writeAndRead(key);
                }
                //处理之后把事件移除
                iter.remove();
                System.out.println("移除key");

            }

        }

    }

    public void connection(SelectionKey key) throws Exception {
        sc.finishConnect();
        sc.register(selector, SelectionKey.OP_WRITE|SelectionKey.OP_READ);
        System.out.println("服务器已经连接上。。。");
    }

    

    public void writeAndRead(SelectionKey key) throws Exception {
        //处理读取事件（从服务端读信息）
        if(key.isReadable()&&switchOver){
            SocketChannel sc = (SocketChannel)key.channel();
            buffer.clear();
            sc.read(buffer);
            buffer.flip();
            byte[] byteM = new byte[buffer.limit()];
            buffer.get(byteM);
            String message = new String(byteM);
            System.out.println("来自服务器的消息："+message);
            switchOver = false;
        }
        //处理写入事件（往服务端写入信息）
        else if(key.isWritable()&&!switchOver){
            Scanner scanner = new Scanner(System.in);
            System.out.println("input <exit> for quit");
            String message = scanner.nextLine();
            ByteBuffer writeBuffer = ByteBuffer.wrap(message.getBytes());
            sc.write(writeBuffer);
            if("exit".equals(message)){
                polling=false;
            }
            switchOver = true;
        }
    }

}

以上的非阻塞NIO示例，设定buffer为128，如果输入超过会报错，对此可以在服务端和客户端加一个Map<SocketChannel, byte[]> clientMessage = new ConcurrentHashMap<>()来存储,只要循环把每次的数据先读到buffer再把buffer中的数据放到map的byte中，map中的byte[]其实就是放一次传递的所有信息，因为buffer一次放不下。