NIO编程

NIO被称为非阻塞IO，它跟BIO不同的地方在于，它如果没有接收到客户端消息的话，可以不阻塞当前服务线程，从而使当前服务线程去继续接收其他客户端线程的请求。

NIO的三大核心组件

1. ByteBuffer
2. ServerSocketChannel
3. Selector

ByteBuffer的使用

//构建一个byte字节缓冲区，默认分配的是堆内存，容量为4，则此时它的limit也为4
        ByteBuffer byteBuffer = ByteBuffer.allocate(4);

//        ByteBuffer byteBuffer = ByteBuffer.allocateDirect(4);//申请堆外内存

        //ByteBuffer的三个重要属性
        System.out.println(String.format("初始化：capacity容量：%s, position位置：%s, limit限制：%s", byteBuffer.capacity(),
                byteBuffer.position(), byteBuffer.limit()));
        //写入三个字节的byte整数
        byteBuffer.put((byte) 1);
        byteBuffer.put((byte) 2);
        byteBuffer.put((byte) 3);
        System.out.println(String.format("写入3个字节后的属性情况：capacity容量：%s, position位置：%s, limit限制：%s", byteBuffer.capacity(),
                byteBuffer.position(), byteBuffer.limit()));

        //读取元素时，需要调用flip()方法将下表重新回到0
        System.out.println("现在开始读取元素");
        byteBuffer.flip();
        byte firstByte = byteBuffer.get();
        System.out.println("读取的第一个元素是：" + firstByte);
        byte secondByte = byteBuffer.get();
        System.out.println("读取的第二个元素是：" + secondByte);
        //现在属于读模式的情况，现在已经存放了三个元素，所以只能读到第三个元素，它的下标是2；所以不能读取到的下标limit=3
        System.out.println(String.format("读取2字节数据后，capacity容量：%s, position位置：%s, limit限制：%s", byteBuffer.capacity(),
                byteBuffer.position(), byteBuffer.limit()));

        //继续对byteBuffer进行写操作
        //此时读模式下，limit=3，position=2.继续写入只能覆盖写入一条数据
        // clear()方法清除整个缓冲区。compact()方法仅清除已阅读的数据。转为写入模式
        System.out.println("切换为写入模式（使用compact方法清除掉已经读取的数据）");
        System.out.println("写入前buffer的内容为：");
        for (int i = 0; i < byteBuffer.limit(); i++) {
            System.out.print(byteBuffer.get(i) + " ");
        }
        byteBuffer.compact();

        byteBuffer.put((byte) 4);
        byteBuffer.put((byte) 5);

        System.out.println();
        System.out.println(String.format("写入2字节数据后，capacity容量：%s, position位置：%s, limit限制：%s", byteBuffer.capacity(),
                byteBuffer.position(), byteBuffer.limit()));

        System.out.println("写入后buffer的内容为：");
        for (int i = 0; i < byteBuffer.limit(); i++) {
            System.out.print(byteBuffer.get(i) + " ");
        }

        System.out.println();
        //读取里边的元素
        byteBuffer.flip();
        System.out.println("再次切换到读模式");
        System.out.println(String.format("当前byteBuffer，capacity容量：%s, position位置：%s, limit限制：%s", byteBuffer.capacity(),
                byteBuffer.position(), byteBuffer.limit()));
        System.out.print("当前的数据为：" );
        for (int i = 0; i < byteBuffer.limit(); i++) {
            System.out.print(byteBuffer.get(i) + " ");
        }

ByteBuffer为性能关键型代码提供了直接内存(direct对外)和非直接内存(heap堆)两种实现。
使用堆外内存的好处：

1. 进行网络IO或者文件IO时比堆内存少一次拷贝。正常的堆内对象，GC时会移动对象内存，在写file或socket的过程中，会先把数据复制到堆外，再进行写入。
2. 直接内存在GC范围之外，减低GC压力，但实现了自动管理。DirectByteBuffer中有一个Cleaner对象(PhantomReference),Cleaner被GC前会执行clean方法，触发DirectByteBuffer中定义的Deallocator。

使用场合：

1. 性能确实可观的时候才去使用；分配给大型、长寿命对象；比如网络传输、文件读写场景
2. 通过虚拟机参数MaxDirectMomorySize限制大小，防止耗尽整个机器的内存

channel通道

相对于BIO来说，NIO编程中不需要以byte数组为数据存储来使用InputStream和OutPutStream两个对象来作为数据的传输对象；NIO编程中可以使用channel对象来进行网络的连接并且以非阻塞的方式通过ByteBuffer为数据载体使用channel来作为传输通道；
channel分为服务端的ServerSocketChannel和客户端的SocketChannel。
客户端使用如下：

/**
	客户端SocketChannel的使用
*/
public class NIOClient {
    public static void main(String[] args) throws Exception {
    	//获取通道
        SocketChannel socketChannel = SocketChannel.open();
        //设置是否为阻塞IO
        socketChannel.configureBlocking(false);
        //连接端口
        socketChannel.connect(new InetSocketAddress("127.0.0.1", 8080));

        while (!socketChannel.finishConnect()) {
            // 没连接上,则一直等待
            Thread.yield();
        }
        Scanner scanner = new Scanner(System.in);
        System.out.println("请输入：");
        // 发送内容
        String msg = scanner.nextLine();
        ByteBuffer buffer = ByteBuffer.wrap(msg.getBytes());
        while (buffer.hasRemaining()) {
            socketChannel.write(buffer);
        }
        // 读取响应
        System.out.println("收到服务端响应:");
        ByteBuffer requestBuffer = ByteBuffer.allocate(1024);

        while (socketChannel.isOpen() && socketChannel.read(requestBuffer) != -1) {
            // 长连接情况下,需要手动判断数据有没有读取结束 (此处做一个简单的判断: 超过0字节就认为请求结束了)
            if (requestBuffer.position() > 0) break;
        }
        requestBuffer.flip();
        byte[] content = new byte[requestBuffer.limit()];
        requestBuffer.get(content);
        System.out.println(new String(content));
        scanner.close();
        socketChannel.close();
    }

}

服务器端由于涉及到Selector的使用，所以我们先来了解一下Selector。

Selector选择器

Selector可以检查一个或多个NIO通道，并确定哪些通道已经准备好进行读取或者写入。实现单个线程可以管理多个通道，从而管理多个网络连接。
一个线程使用Selector监听多个channel的不同事件，它包含四类事件，对应SelectionKey的四个常量：

1. Connect连接(SelectionKey.OP_CONNECT)
2. Accept准备就绪(SelectionKey.OP_ACCEPT)
3. Read读取(SelectionKey.OP_READ)
4. Write写入(SelectionKey.OP_WRITE)

实现一个线程处理多个通道的核心理念：事件驱动机制。非阻塞的网络通道下，开发者通过Selector注册对于通道感兴趣的事件类型，线程通过监听事件来触发相应的代码执行。(更底层是操作系统的多路复用机制)

它的使用如下：

/**
 * 结合selector实现非阻塞的服务端，借助消息机制放弃对channel的轮训
 */
public class NIOServer {

    public static void main(String[] args) throws IOException {
        // 1. 创建网络服务端ServerSocketChannel
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        //设置为非阻塞IO方式
        serverSocketChannel.configureBlocking(false);

        //2.构建一个Selector，将channel注册上去
        Selector selector = Selector.open();
        // 将serverSocketChannel注册到selector
        SelectionKey selectionKey = serverSocketChannel.register(selector, 0, serverSocketChannel);
        //对serverSocketChannel上面的accept事件感兴趣(serverSocketChannel只能支持accept操作)
        selectionKey.interestOps(SelectionKey.OP_ACCEPT);

        //3.绑定服务器端口
        serverSocketChannel.socket().bind(new InetSocketAddress(8000));
        System.out.println("服务器启动");

        //开始轮询
        while(true){
            //使用selector.select方法进行轮询,该方法有阻塞效果,直到有事件通知才会有返回
            selector.select();
            //获取所有的注册事件key
            Set<SelectionKey> selectionKeys = selector.selectedKeys();
            //遍历这些注册事件的key
            Iterator<SelectionKey> keyIterator = selectionKeys.iterator();

            while (keyIterator.hasNext()){
                SelectionKey currentSelectionKey = keyIterator.next();
                keyIterator.remove();

                //关注read和accept这两类注册事件
                if(currentSelectionKey.isAcceptable()){
                    //获取当前key对应的事件对象
                    ServerSocketChannel current_serverSocketChannel = (ServerSocketChannel) currentSelectionKey.attachment();
                    //来获取socketChannel,将拿到的客户端连接通道,注册到selector上面
                    //mainReactor
                    SocketChannel clientSocketChannel = current_serverSocketChannel.accept();
                    clientSocketChannel.configureBlocking(false);
                    clientSocketChannel.register(selector,SelectionKey.OP_READ,clientSocketChannel);
                    System.out.println("收到新连接 : " + clientSocketChannel.getRemoteAddress());
                }

                //判断是否有数据可读
                if(currentSelectionKey.isReadable()){
                    SocketChannel socketChannel = (SocketChannel) currentSelectionKey.attachment();
                    try{
                        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                        while (socketChannel.isOpen() && socketChannel.read(byteBuffer)!= -1){
                            // 长连接情况下,需要手动判断数据有没有读取结束 (此处做一个简单的判断: 超过0字节就认为请求结束了)
                            if (byteBuffer.position() > 0) {break;}
                        }
                        //如果没有数据，则进行其他注册事件的处理
                        if(byteBuffer.position() ==0) {continue;}
                        //如果客户端返回有数据，则进行数据读取
                        byteBuffer.flip();
                        //
                        byte[] bytesContent = new byte[byteBuffer.limit()];
                        byteBuffer.get(bytesContent);
                        System.out.println(socketChannel.getRemoteAddress() +"服务器,收到的数据是：" + String.valueOf(bytesContent));
                        // TODO 业务操作 数据库 接口调用等等

                        // 响应结果 200
                        String response = "HTTP/1.1 200 OK\r\n" +
                                "Content-Length: 11\r\n\r\n" +
                                "hello,i am server";
                        ByteBuffer output_buffer = ByteBuffer.wrap(response.getBytes());
                        while (output_buffer.hasRemaining()){
                            socketChannel.write(output_buffer);
                        }
                    }catch (IOException ioException){
                        //出现异常则取消该事件
                        currentSelectionKey.cancel();
                    }
                }
            }
            selector.selectNow();
        }
    }
}

注意：上边代码中一个selector监听所有事件,一个线程处理所有请求事件会成为瓶颈，并且不能充分发挥多核CPU的作用，要结合多线程的使用来处理。这里我们可以参考Doug Lea的《Scalable IO in Java》来把NIO和多线程技术结合起来。

多路复用的Reactor线程模型

如果服务端只使用一个线程来处理所有请求的话，就不能够高效的使用到CPU，所以我们结合NIO和多线程技术来实现一种Reactor的线程模型。
代码如下：

/**
 * NIO selector 多路复用reactor线程模型
 */
public class NIOServer_Reactor {
    /** 处理业务操作的线程 */
    private static ExecutorService workPool = Executors.newCachedThreadPool();

    /**
     * 封装了selector.select()等事件轮询的代码
     */
    abstract class ReactorThread extends Thread {

        Selector selector;
        LinkedBlockingQueue<Runnable> taskQueue = new LinkedBlockingQueue<>();

        /**
         * Selector监听到有事件后,调用这个方法
         */
        public abstract void handler(SelectableChannel channel) throws Exception;

        private ReactorThread() throws IOException {
            selector = Selector.open();
        }

        volatile boolean running = false;

        @Override
        public void run() {
            // 轮询Selector事件
            while (running) {
                try {
                    // 执行队列中的任务
                    Runnable task;
                    while ((task = taskQueue.poll()) != null) {
                        task.run();
                    }
                    selector.select(1000);

                    // 获取查询结果
                    Set<SelectionKey> selected = selector.selectedKeys();
                    // 遍历查询结果
                    Iterator<SelectionKey> iter = selected.iterator();
                    while (iter.hasNext()) {
                        // 被封装的查询结果
                        SelectionKey key = iter.next();
                        iter.remove();
                        int readyOps = key.readyOps();
                        // 关注 Read 和 Accept两个事件
                        if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
                            try {
                                SelectableChannel channel = (SelectableChannel) key.attachment();
                                channel.configureBlocking(false);
                                handler(channel);
                                if (!channel.isOpen()) {
                                    key.cancel(); // 如果关闭了,就取消这个KEY的订阅
                                }
                            } catch (Exception ex) {
                                key.cancel(); // 如果有异常,就取消这个KEY的订阅
                            }
                        }
                    }
                    selector.selectNow();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

        private SelectionKey register(SelectableChannel channel) throws Exception {
            // 为什么register要以任务提交的形式，让reactor线程去处理？
            // 因为线程在执行channel注册到selector的过程中，会和调用selector.select()方法的线程争用同一把锁
            // 而select()方法是在eventLoop中通过while循环调用的，争抢的可能性很高，为了让register能更快的执行，就放到同一个线程来处理
            FutureTask<SelectionKey> futureTask = new FutureTask<>(() -> channel.register(selector, 0, channel));
            taskQueue.add(futureTask);
            return futureTask.get();
        }

        private void doStart() {
            if (!running) {
                running = true;
                start();
            }
        }
    }

    private ServerSocketChannel serverSocketChannel;
    // 1、创建多个线程 - accept处理reactor线程 (accept线程)
    private ReactorThread[] mainReactorThreads = new ReactorThread[1];
    // 2、创建多个线程 - io处理reactor线程  (I/O线程)
    private ReactorThread[] subReactorThreads = new ReactorThread[8];

    /**
     * 初始化线程组
     */
    private void newGroup() throws IOException {
        // 创建IO线程,负责处理客户端连接以后socketChannel的IO读写
        for (int i = 0; i < subReactorThreads.length; i++) {
            subReactorThreads[i] = new ReactorThread() {
                @Override
                public void handler(SelectableChannel channel) throws IOException {
                    // work线程只负责处理IO处理，不处理accept事件
                    SocketChannel ch = (SocketChannel) channel;
                    ByteBuffer requestBuffer = ByteBuffer.allocate(1024);
                    while (ch.isOpen() && ch.read(requestBuffer) != -1) {
                        // 长连接情况下,需要手动判断数据有没有读取结束 (此处做一个简单的判断: 超过0字节就认为请求结束了)
                        if (requestBuffer.position() > 0) {break;}
                    }
                    if (requestBuffer.position() == 0) {return;} // 如果没数据了, 则不继续后面的处理
                    requestBuffer.flip();
                    byte[] content = new byte[requestBuffer.limit()];
                    requestBuffer.get(content);
                    System.out.println(new String(content));
                    System.out.println(Thread.currentThread().getName() + "收到数据,来自：" + ch.getRemoteAddress());

                    // TODO 业务操作 数据库、接口...
                    workPool.submit(() -> {
                    });

                    // 响应结果 200
                    String response = "HTTP/1.1 200 OK\r\n" +
                            "Content-Length: 11\r\n\r\n" +
                            "Hello World";
                    ByteBuffer buffer = ByteBuffer.wrap(response.getBytes());
                    while (buffer.hasRemaining()) {
                        ch.write(buffer);
                    }
                }
            };
        }

        // 创建mainReactor线程, 只负责处理serverSocketChannel
        for (int i = 0; i < mainReactorThreads.length; i++) {
            mainReactorThreads[i] = new ReactorThread() {
                AtomicInteger incr = new AtomicInteger(0);

                @Override
                public void handler(SelectableChannel channel) throws Exception {
                    // 只做请求分发，不做具体的数据读取
                    ServerSocketChannel ch = (ServerSocketChannel) channel;
                    SocketChannel socketChannel = ch.accept();
                    socketChannel.configureBlocking(false);
                    // 收到连接建立的通知之后，分发给I/O线程继续去读取数据
                    int index = incr.getAndIncrement() % subReactorThreads.length;
                    ReactorThread workEventLoop = subReactorThreads[index];
                    workEventLoop.doStart();
                    SelectionKey selectionKey = workEventLoop.register(socketChannel);
                    selectionKey.interestOps(SelectionKey.OP_READ);
                    System.out.println(Thread.currentThread().getName() + "收到新连接 : " + socketChannel.getRemoteAddress());
                }
            };
        }


    }

    /**
     * 初始化channel,并且绑定一个eventLoop线程
     *
     * @throws IOException IO异常
     */
    private void initAndRegister() throws Exception {
        // 1、 创建ServerSocketChannel
        serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
        // 2、 将serverSocketChannel注册到selector
        int index = new Random().nextInt(mainReactorThreads.length);
        mainReactorThreads[index].doStart();
        SelectionKey selectionKey = mainReactorThreads[index].register(serverSocketChannel);
        selectionKey.interestOps(SelectionKey.OP_ACCEPT);
    }

    /**
     * 绑定端口
     *
     * @throws IOException IO异常
     */
    private void bind() throws IOException {
        //  1、 正式绑定端口，对外服务
        serverSocketChannel.bind(new InetSocketAddress(8000));
        System.out.println("启动完成，端口8000");
    }

    public static void main(String[] args) throws Exception {
        NIOServer_Reactor nioServer_Reactor = new NIOServer_Reactor();
        nioServer_Reactor.newGroup(); // 1、 创建main和sub两组线程
        nioServer_Reactor.initAndRegister(); // 2、 创建serverSocketChannel，注册到mainReactor线程上的selector上
        nioServer_Reactor.bind(); // 3、 为serverSocketChannel绑定端口
    }
}