JavaNIO与Reactor模式

一、NIO的简单介绍 

      Java NIO(New IO)是一个可以替代标准Java IO API的IO API（从Java 1.4开始)，Java NIO提供了与标准IO不同的IO工作方式。 NIO中的核心内容有Channel、Buffer、Selector,其他组件如Pipe和FileLock只不过是三个组件共同使用的工具类。

     1、Channle

          Channle的实现主要有FileChannel、DatagramChannel、SocketChannle、ServerSocketChannel。这些通道覆盖了文件IO、UDP和TCP网络IO

     2、Buffer

          Buffer的实现由ByteBuf、CharBuffer、DoubleBuffer、FloatBuffer、IntBuffer、LongBuffer、ShortBuffer以及MappedByteBuffer(用于表示内存映射文件)。以上Buffer涵盖了能够通过IO发送的基本数据类型。

     3、Selector

       Selector允许单线程处理多个 Channel。如果你的应用打开了多个连接（通道），但每个连接的流量都很低，使用Selector就会很方便。Selector（选择器）是Java NIO中能够检测一到多个NIO通道，并能够知晓通道是否为诸如读写事件做好准备的组件。这样，一个单独的线程可以管理多个channel，从而管理多个网络连接。

   二、NIO在网络服务端的应用。

      2.1、传统ServerSocket

       当我们刚开始学习想要做一个聊天服务器的时候，最简单的做法就是使用ServerSocket来作为服务端，接收Socket客户端发送来的数据。下面的代码就是一个简单的 例子，当我们从服务端接收到数据时，新建一个线程来处理我们的数据，这种模式是一种阻塞式IO,在高并发的情况下，我们需要新建大量线程，然而程序的效率在线程数不多的情况下是随着线程的增加而增加，但是到达一定数量后是随着线程的增加而减少。所以传统阻塞式IO的瓶颈在于不能处理过多的连接。

    2.2、NIO非阻塞式服务。

       javaNIO中的ServerSocketChannel是一个可以监听新进来的TCP连接的通道，就像标准的ServerSocket一样。我们在这里使用NIO来搭建一个服务端。

    

package NIO;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.Buffer;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;

public class NIOServer {

    public static  void main(String[] args)throws IOException{
        //创建一个NIOsocket服务
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        //将ServerSocketChannel设置为非阻塞
        serverSocketChannel.configureBlocking(false);
        //serverSocketChannel.bind(new InetSocketAddress(8989));
        serverSocketChannel.socket().bind(new InetSocketAddress(8989));
        //调用 Selector.open()方法创建一个selector
        Selector selector = Selector.open();
        //将serverSocketChannel注册到selector中，监听TCP连接
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        while(true){
            //这条语句会阻塞
            int nkey = selector.select();
           //也可以设置超时时间，防止进程阻塞
          //int nkey = selector.select(long timeout);
            System.out.println("nkey"+nkey);
            if(nkey>0) {
                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                int i = 0;
                while (iterator.hasNext()) {
                    i++;
                    System.out.println("iteratorLength" + i);
                    SelectionKey key = iterator.next();
                    iterator.remove();
                    reactor(key, selector);
                }
            }
        }
    }

    public static void reactor(SelectionKey key,Selector selector) throws IOException {
        if (key.isAcceptable()) {
            //接收就绪状态
            handlerAccept(key,selector);
        } else if (key.isReadable()) {
            //读就绪状态
            handlerReader(key);
        }
    }

    public static void handlerAccept(SelectionKey key,Selector selector) throws IOException {
        ServerSocketChannel sever = (ServerSocketChannel) key.channel();
        SocketChannel channel = sever.accept();
        channel.configureBlocking(false);
        System.out.println("客户端连接" + channel.toString());
        //将通道注册到selector中
        //此时Selector监听channel并对read感兴趣
        channel.register(selector, SelectionKey.OP_READ);
    }

    public  static void handlerReader(SelectionKey key) throws IOException {

        SocketChannel socketChannel = (SocketChannel) key.channel();
        //读取部分消息，不确定数据长度只用了1024个字节长度接收，当数据大于1024时，会丢失数据，当数据只有几十个字节时会浪费内存空间，当并发数据量大时会造成不必要的浪费
        // 文章 http://ifeve.com/non-blocking-server/
        //  http://tutorials.jenkov.com/java-performance/resizable-array.html
        // 中介绍了如何实现这样支持可调整大小的数组的内存缓冲区
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        //不会阻塞
        int n =0;

        //此处需要捕获异常，否则若客户端强制关闭，服务器会报“Java.io.IOException: 远程主机强迫关闭了一个现有的连接。”，
        // 并且服务器会在报错后停止运行，错误的意思就是客户端关闭了，但是服务器还在从这个套接字通道读取数据，便抛出IOException，
        // 导致这种情况出现的原因就是，客户端异常关闭后，服务器的选择器会获取到与客户端套接字对应的套接字通道SelectionKey，
        // 并且这个key的兴趣是OP_READ，执行从这个通道读取数据时，客户端已套接字已关闭，所以会出现“java.io.IOException: 远程主机强迫关闭了一个现有的连接”的错误。
        // 所以在服务器在读取数据时，若发生异常，则取消当前key并关闭通道，如下代码：
        try {
            n = socketChannel.read(buffer);
        }catch (Exception e){
            e.printStackTrace();
            key.cancel();
            socketChannel.socket().close();
            socketChannel.close();
            return;
        }
        System.out.println("读取字节数"+n);
        if (n > 0) {
            byte[] data = buffer.array();
            System.out.println("服务端收到信息:" + new String(data, 0, n));
            buffer.flip();
            socketChannel.write(buffer);
        } else {
            System.out.println("clinet is close");
            key.cancel();
        }

    }
}

       上面是一个非阻塞式IO的服务端。其设计是一个经典的单线程的reactor模式。

三、Reactor模式

 

 

     Reactor不仅监控所有的连接请求，还监控其他读请求。我们将多个channle注册到selector中，其中serverSocketChannle只注册了OP_ACCEPT事件，而socketChannel只注册了OP_READ事件。

       Select调用select()方法后，检测所有注册的通道，返回给Reactor(反应器)我们感兴趣的事件已经准备就绪的那些通道，然后Reactor分发给不同的handler处理(handlerAccept、handlerReader)。