Netty权威指南——基础篇2（NIO编程）备份

1 概述

        与Socket类和ServerSocket，NIO也提供了SocketChannel和ServerSocketChannel两种不同的套接字通道实现。这两种新增的通道都支持阻塞和非阻塞两种模式。阻塞模式使用简单，但性能和可靠性都不好，非阻塞模式则正好相反。一般来说，低负载、地并发的应用程序可以选择同步阻塞I/O降低变成复杂度；对于高负载、高并发的网络应用，需要使用NIO的非阻塞模型进行开发。

2 NIO类库简介

        新的输入/输出（NIO）库是在JDK 1.4中引入的。NIO弥补了原来同步阻塞I/O的不足，它在标准java代码中提供了高速、面向块的I/O。通过定义包含数据的类，以及通过以块的形式处理这些数据，NIO不用使用本机代码就可以利用低级优化，这是原来的I/O无法做到的。

2.1 缓冲区Buffer

        首先介绍缓冲区（Buffer）的概念。Buffer是一个对象，它包含一些要写入或者要读出的数据。在NIO类库中加入Buffer对象，体现了新库与原I/O的一个重要区别。在面向流的I/O中，可以将数据直接写入或者将数据直接读到Stream对象中。

        在NIO库中，所有数据都是用缓冲区处理的。在读取数据时，它是直接读到缓冲区中；在写入数据时，也是写入到缓冲区中。任何使用访问NIO中的数据，都是通过缓冲区进行操作。

        缓冲区实质上是一个数组。通常它是一个字节数组（ByteBuffer），也可以使用其他类型的数据。但是一个缓冲区不仅仅是一个数组，缓冲区提供了对数据的结构化访问以及维护读写位置等信息。

        最常用的缓冲区是ByteBuffer，一个ByteBuffer提供了一组功能用于操作byte数据。除了ByteBuffer，还有其他的一些 缓冲区，实际上，每一种java基本类型（除了boolean类型）都对应有一种缓冲区，具体如下：

        1、ByteBuffer：字节缓冲区

        2、CharBuffer：字符缓冲区

        3、ShortBuffer：短整型缓冲区

        4、IntBuffer：整型缓冲区

        5、LongBuffer：长整型缓冲区

        6、FloatBuffer：浮点型缓冲区

        7、DoubleBuffer：双精度浮点型缓冲区

        缓冲区的类图继承关系如下：

        每一个Buffer类都是Buffer接口的一个子实例。除了ByteBuffer，每一个Buffer类都有完全一样的操作，只是它们所处理的数据类型不一样。因为大多数标准I/O操作都是用ByteBuffer，所以它在具有一般缓冲区的操作之外还提供了一些特有的操作，以方便网络读写。

2.2 通道Channel

        Channel是一个通道，就像自来水管一样，网络数据通过Channel读取和写入。通道与流的不同至于在于通道时双向的，流只是在一个方向上移动（一个流必须是InputStream或OutputStream的子类），而通道可用于读、写或者二者同时进行。

        因为Channel是全双工的，所以它可以比流更好的映射底层操作系统的API。特别是在UNIX网络编程模型中，底层操作系统的通道全都是全双攻的，同时支持读写操作。

        Channel的类图继承关系如下：

 

        自顶向下看，前三层主要是Channel接口，用于定义它的功能，后面是一些具体的功能类（抽象类）。从图中可以看出，实际上Channel分为两大类：用于王立国读写的SelectableChannel和用于文件操作的FileChannel。 

2.3 多路复用器Selector

        多路复用器Selector，它是Java NIO编程的基础，熟练地掌握Selector对于NIO编程至关重要。多路复用器提供选择已经就绪的任务的能力。简单来说，Selector会不断地轮询注册在其上的Channel，如果某个Channel上面发生读写事件，这个Channel就处于就绪状态，会被Selector轮询出来，然后通过SelectionKey可以获取就绪Channel的集合，记性后续的I/O操作。

        一个多路复用器Selector可以同时轮询多个Channel，由于JDK使用了epoll代替传统的select实现，所以它并没有最大连接句柄1024/2048的限制。这样就意味着只需要一个Selector线程负责Selector的轮询，就可以接入成千上万个客户端。

3 NIO服务端序列图

        NIO服务端通信序列图如下：

        第一步：打开ServerSocketChannel，用于监听客户端的连接，它是所有客户端连接的父管道，示例代码如下：

ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();

         第二步：绑定监听端口，设置连接为非阻塞模型，示例代码如下：

int port = 8080;
serverSocketChannel.socket().bind(new InetSocketAddress(InetAddress.getByName("IP"), port));
serverSocketChannel.configureBlocking(false);

        第三步：创建Reactor线程，创建多路复用器并启动线程，示例代码如下：

//创建Reactor线程
new Thread(new ReactorTask()).start();
//创建多路复用器
Selector selector = Selector.open();

        第四步：将ServerSocketChannel注册到Reactor线程的多路复用器Selector上，监听ACCEPT事件，示例代码如下：

SelectionKey key = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT, ioHandler);

        第五步：多路复用器在线程run()方法的无限循环体内轮询准备就绪的key，示例代码如下：

        int num = selector.select();
        Set<SelectionKey> selectededKeys = selector.selectedKeys();
        Iterator<SelectionKey> iterator = selectededKeys.iterator();
        while(iterator.hasNext()){
            SelectionKey key = (SelectionKey) iterator.next();
            //do something
        }

        第六步：多路复用器监听到有新的客户端接入，处理新的接入请求，完成TCP三次握手，建立物理链路，示例代码如下：

SocketChannel channel = serverSocketChannel.accept();

        第七步：设置客户端链路为非阻塞模式，示例代码如下：

        channel.configureBlocking(false);
        channel.socket().setReuseAddress(true);

        第八步：将新接入的客户端连接注册到Reactor线程的多路复用器上，监听读操作，读取客户端发送的网络消息，示例代码如下：

SelectionKey selectionKey = channel.register(selector, SelectionKey.OP_READ);

        第九步：异步读取客户端请求消息到缓冲区，示例代码如下：

long read = channel.read(new ByteBuffer[1024]);

        第十步：对ByteBuffer进行编解码，如果有半包消息指针reset，继续读取后续的报文，将解码成功的消息封装成Task，投递到业务线程池中，进行业务逻辑编排，示例代码如下：

        List<Object> messageList = new ArrayList<>();
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        while(buffer.hasRemaining()){
            buffer.mark();
            Object message = decode(buffer);
            if(message == null){
                buffer.reset();
                break;
            }
            messageList.add(message);
        }
        if(!buffer.hasRemaining()){
            buffer.clear();
        }else{
            buffer.compact();

        }
        if(messageList != null && ! messageList.isEmpty()){
            for(Object m : messageList){
                handlerTask(m);
            }
        }

        第十一步：将POJO对象encode成ByteBuffer，调用SocketChannel的异步write方法，将消息异步发送给客户端，示例代码如下：

        channel.write(buffer);

        【注意】如果发送区TCP缓冲区满，会导致写半包，此时，需要注册监听写操作位，循环写，直到整包消息写入TCP缓冲区。

        以上11步就是NIO服务器序列，整体示例代码如下：（不要运行，仅供参考）

public class NIOStep {
    public static void main(String[] args) throws IOException {
        //创建ServerSocketChannel，用于监听客户端连接
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        //端口号
        int port = 8080;
        //设置IP地址和端口号
        serverSocketChannel.socket().bind(new InetSocketAddress(InetAddress.getByName("IP"), port));
        //设置连接为非阻塞模式
        serverSocketChannel.configureBlocking(false);
        //创建Reactor线程
        new Thread(new ReactorTask()).start();
        //创建多路复用器
        Selector selector = Selector.open();
        //将ServerSocketChannel注册到Reactor线程的多路复用器Selector上，监听ACCEPT事件
        SelectionKey keys = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT, ioHandler);
        //Selector在run方法内轮询准备就绪的key
        int num = selector.select();
        Set<SelectionKey> selectededKeys = selector.selectedKeys();
        Iterator<SelectionKey> iterator = selectededKeys.iterator();
        while(iterator.hasNext()){
            SelectionKey key = (SelectionKey) iterator.next();
            //do something
        }
        SocketChannel channel = serverSocketChannel.accept();
        channel.configureBlocking(false);
        channel.socket().setReuseAddress(true);
        SelectionKey selectionKey = channel.register(selector, SelectionKey.OP_READ);

        long read = channel.read(new ByteBuffer[1024]);

        List<Object> messageList = new ArrayList<>();
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        while(buffer.hasRemaining()){
            buffer.mark();
            Object message = decode(buffer);
            if(message == null){
                buffer.reset();
                break;
            }
            messageList.add(message);
        }
        if(!buffer.hasRemaining()){
            buffer.clear();
        }else{
            buffer.compact();

        }
        if(messageList != null && ! messageList.isEmpty()){
            for(Object m : messageList){
                handlerTask(m);
            }
        }
        channel.write(buffer);
    }
}

4 NIO创建TimeServer

        TimeServer.java 代码

public class TimeServer {
    public static void main(String[] args) {
        int port = 8080;
        MultiplexerTimeServer timeServer = new MultiplexerTimeServer(port);
        new Thread(timeServer,"NIO-MultiplexerTimeServer-001").start();
    }
}

        其中，

MultiplexerTimeServer timeServer = new MultiplexerTimeServer(port);

       创建了一个被称为 MultiplexerTimeServer的多路复用类，它是一个独立的线程，负责轮询多路复用器Selector，可以处理多个客户端的并发接入。

MultiplexerTimeServer.java 代码

package com.jay.NIOBASIC;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Date;
import java.util.Iterator;
import java.util.Set;

public class MultiplexerTimeServer implements Runnable{
    private Selector selector;
    private ServerSocketChannel serverSocketChannel;
    private volatile  boolean stop;

    //初始化多路复用器，绑定监听端口
    public MultiplexerTimeServer(int port){
        try {
            selector = Selector.open();
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.configureBlocking(false);
            serverSocketChannel.socket().bind(new InetSocketAddress(port),1024);
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            System.out.println("TimeServer 开始启动，端口是:"+port);
        }catch (IOException e){
            e.printStackTrace();
            System.exit(1);
        }
    }
    public void stop(){
        this.stop = true;
    }

    @Override
    public void run(){
        while(!stop){
            try {
                selector.select(1000);
                Set<SelectionKey> selectededKeys = selector.selectedKeys();
                Iterator<SelectionKey> it = selectededKeys.iterator();
                SelectionKey key = null;
                while(it.hasNext()){
                    key = it.next();
                    it.remove();
                    try {
                        handlerInput(key);
                    }catch (Exception e){
                        if(key != null){
                            key.cancel();
                            if(key.channel() != null){
                                key.channel().close();
                            }
                        }
                    }
                }
            }catch (Throwable t){
                t.printStackTrace();
            }
        }
        //多路复用器关闭后，所有注册到上面的Channel和Pipe等资源都会被自动取注册并关闭，所有不需要重复释放资源
        if(selector != null){
            try {
                selector.close();
            }catch (IOException e){
                e.printStackTrace();
            }
        }
    }

    private void handlerInput(SelectionKey key) throws IOException{
        if(key.isValid()){
            //处理新接入的请求消息
            if(key.isAcceptable()){
                ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                SocketChannel sc = ssc.accept();
                sc.configureBlocking(false);
                sc.register(selector,SelectionKey.OP_READ);
            }
            //处理读取的请求消息
            if(key.isReadable()){
                SocketChannel sc = (SocketChannel) key.channel();
                ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                int readBytes = sc.read(readBuffer);
                if(readBytes > 0){
                    readBuffer.flip();
                    byte[] bytes = new byte[readBuffer.remaining()];
                    readBuffer.get(bytes);
                    String body = new String(bytes,"UTF-8");
                    System.out.println("TimeServer 接收:"+body);
                    String currentTime = "query".equalsIgnoreCase(body) ? new Date(System.currentTimeMillis()).toString() : "bad";
                    doWrite(sc,currentTime);
                    
                }else if(readBytes < 0){
                    //对端链路关闭
                    key.cancel();
                    sc.close();
                }else{
                    //读到0字节，忽略
                }
            }
        }
    }
    private void doWrite(SocketChannel channel,String response) throws IOException {
        if(response != null && response.trim().length() > 0){
            byte[] bytes = response.getBytes();
            ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
            writeBuffer.put(bytes);
            writeBuffer.flip();
            channel.write(writeBuffer);
        }
    }
}

        代码分析：

        1、MultiplexerTimeServer()是构造方法，在构造方法中进行资源初始化。创建多路复用器Selector、ServerSocketChannel，对Channel和TCP参数进行配置。系统资源初始化成功后，将ServerSocketChannel注册到Selector，监听SelectionKey.OP_ACCEPT操作位。如果初始化失败则退出。

        2、run()方法，在while循环体中循环遍历selector，它的休眠时间为1秒。无论是否有读写等时间发生，selector每个1秒被唤醒一次。selector也提供了一个无参的select方法：当有处于就绪状态的Channel时，selector将返回该Channel的SelectionKey集合。通过对就绪状态的Channel集合进行迭代，可以进行网络的异步读写操作。

        3、handlerInput方法中的

if(key.isAcceptable()){}

        分支处理新接入的客户端请求消息，根据SelectionKey的操作位进行判断即可获取网络事件的类型，通过ServerSocketChannel的accept接收客户端的连接请求并创建SocketChannel实例。完成上述操作后，相当月完成了TCP的三次握手，TCP物理链路正式建立。

        4、handlerInput方法中的

 if(key.isReadable()){}

        分支用于读取客户端的请求消息。首先创建一个ByteBuffer，用于事先无法得知客户端发送的码流大小，作为例程，开辟了一个1MB的缓冲区。然后调用SocketChannel的read方法读取请求码流。注意，由于已经将SocketChannel设置为异步非阻塞模式，因此它的read是非阻塞的，使用返回值进行判断，看读取到的字节数，返回值有以下三种可能的结果：

        （1）返回大于0：读到了字节，对字节进行编解码；

        （2）返回值等于0：没有读取到字节，属于正常场景 ，忽略；

        （3）返回值为-1：链路已经关闭，需要关闭SocketChannel，释放资源。

        当读取到码流以后，进行解码。首先对readBuffer进行flip操作，它的作用是将缓冲区当前的limit设置为position，position设置为0，用于后续对缓冲区的读取操作。然后根据缓冲区可读的字节个数创建字节数组，调用ByteBuffer的get方法将缓冲区可读的字节数组复制到新创建的字节数组中，最后调用字符串的构造函数创建请求体并打印。如果请求指令是“query”，则把服务器的当时时间编码后返回给客户端。

        5 doWrite方法：

        将应答消息异步发送给客户端。首先将字符串编码成字节数组，根据字节数组的容量创建ByteBuffer，调用put方法将字节数组复制到缓冲区中，然后对缓冲区进行flip操作，最后调用SocketChannel的write方法将缓冲区中的字节数组发送出去。需要指出的是，由于SocketChannel是异步非阻塞的，并不保证一次能够把需要发送的字节数组发送完，此时会出现“写半包”问题。需要注册写操作，不断轮询Selector将没有发送完的ByteBuffer发送完毕，然后可以通过hasRemain方法判断消息是否发送完成。此处仅仅是简单的入门级示例，后续会演示如何处理“写半包”的场景。

5 NIO客户端序列图

        

        第一步：打开SocketChannel，绑定客户端本地地址

SocketChannel clientChannel = SocketChannel.open();

         第二步：设置SocketChannel为非阻塞模式，同时设置客户端连接的TCP参数

        Socket socket = new Socket();
        socket.setReuseAddress(true);
        socket.setReceiveBufferSize(BUFFER_SIZE);
        socket.setSendBufferSize(BUFFER_SIZE);

        第三步：异步连接服务端

boolean connect = clientChannel.connect(new InetSocketAddress("127.0.0.1", 8080));

        第四步：判断是否连接成功，如果成功，则直接注册读状态位到多路复用器中，如果失败（异步连接，返回false，说明客户端已经发送sync包，服务度没有返回ack包，物理链路还没有建立）

selector = Selector.open();
        if(connect){
            clientChannel.register(selector, SelectionKey.OP_READ,ioHandler);
        }else{
            clientChannel.register(selector,SelectionKey.OP_CONNECT,ioHandler)
        }

        第五步：向Reactor线程的多路复用器注册OP_CONNECT状态位，监听服务端的TCP ACK应答

clientChannel.register(selector,SelectionKey.OP_CONNECT,ioHandler);

        第六步：创建Reactor线程，创建多路复用器并启动线程

        selector = Selector.open();
        new Thread(new ReactorTask()).start();

        第七步：多路复用器在线程run方法的无限循环体内轮询准备就绪的key

int num = selector.select();
        Set<SelectionKey> selectededKeys = selector.selectedKeys();
        Iterator<SelectionKey> it = selectededKeys.iterator();
        while(it.hasNext()){
            SelectionKey key = (SelectionKey) it.next();
        }

        第八步：接收connect事件进行处理

if(key.isConnectable()){
                //处理
            }

        第九步：判断连接成功，如果成功，注册读事件到多路复用器

if(clientChannel.finishConnect()){
            registerRead();
        }

        第十步：注册读事件到多路复用器

clientChannel.register(selector,SelectionKey.OP_READ,ioHandler);

        第十一步：异步读客户端请求信息到缓冲区

long read = clientChannel.read(new ByteBuffer[1024]);

        第十二步：对ByteBuffer进行编解码，如果有半包消息接收到缓冲区Reset，继续读取后续的报文，将解码成功的消息封装成 Task，投递到业务线程池中，进行业务逻辑编排。

        第十三步：将POJO对象encode成ByteBuffer，调用SocketChannel的异步write接口，将消息异步发送给客户端

6 NIO创建TimeClient

TimeClient.java

public class TimeClient {
    public static void main(String[] args) {
        int port = 8080;
        new Thread(new TimeClientHandler("127.0.0.1",port),"TimeClient-001").start();
    }
}

TimeClientHandler.java

package com.jay.NIOBASIC;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;

public class TimeClientHandler implements Runnable{
    private String host;
    private int port;
    private Selector selector;
    private SocketChannel socketChannel;
    private volatile boolean stop;
    
    public TimeClientHandler(String host,int port){
        this.host = host == null ? "127.0.0.1" : host;
        this.port = port;
        try {
            selector = Selector.open();
            socketChannel = SocketChannel.open();
            socketChannel.configureBlocking(false);
        }catch (IOException e){
            e.printStackTrace();
            System.exit(1);
        }
    }
    
    @Override
    public void run(){
        try {
            doConnect();
        }catch (IOException e){
            e.printStackTrace();
            System.exit(1);
        }
        while(!stop){
            try {
                selector.select(1000);
                Set<SelectionKey> selectededKeys = selector.selectedKeys();
                Iterator<SelectionKey> it = selectededKeys.iterator();
                SelectionKey key = null;
                while(it.hasNext()){
                    key = it.next();
                    it.remove();
                    try {
                        handleInput(key);
                    }catch (Exception e){
                        key.cancel();
                        if(key.channel() != null){
                            key.channel().close();
                        }
                    }
                }
            }catch (Exception e){
                e.printStackTrace();
                System.exit(1);
            }
        }
        if(selector != null){
            try {
                selector.close();
            }catch (IOException e){
                e.printStackTrace();
            }
        }
        
    }
    
    private void handleInput(SelectionKey key) throws IOException{
        if(key.isValid()){
            //判断是否连接成功
            SocketChannel sc = (SocketChannel)key.channel();
            if(key.isConnectable()){
                sc.register(selector,SelectionKey.OP_READ);
                doWrite(sc);
            }else{
                System.exit(1);
            }
            if(key.isReadable()){
                ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                int readBytes = sc.read(readBuffer);
                if(readBytes > 0){
                    readBuffer.flip();
                    byte[] bytes = new byte[readBuffer.remaining()];
                    readBuffer.get(bytes);
                    String body = new String(bytes,"UTF-8");
                    System.out.println("now is : " + body);
                    this.stop = true;
                }else if(readBytes < 0){
                    key.cancel();
                    sc.close();
                }
            }
        }
    }
    private void doConnect() throws IOException{
        //如果直接连接成功，则注册到多路复用器上，发送请求消息，读应答
        if(socketChannel.connect(new InetSocketAddress(host,port))){
            socketChannel.register(selector,SelectionKey.OP_READ);
            doWrite(socketChannel);
        }else{
            socketChannel.register(selector,SelectionKey.OP_CONNECT);
        }
    }
    
   private void doWrite(SocketChannel sc) throws IOException{
        byte[] req = "query".getBytes();
       ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);
       writeBuffer.put(req);
       writeBuffer.flip();
       sc.write(writeBuffer);
       if(!writeBuffer.hasRemaining()){
           System.out.println("发送 2 服务成功");
       }
   } 
}

        代码分析：

        1、构造方法：用于初始化NIO的多路复用器和SocketChannel对象。需要注意的是，创建SocketChannel之后，需要将其设置为异步非阻塞模式。

        2、run()方法中的doConnect方法用于发送连接请求，作为实例，连接是成功的，所以不需要做重连操作，因此将其放到循环之前。下面看doConnect方法实现，首先对SocketChannel的connect()操作进行判断。如果连接成功，则将SocketChannel注册到多路复用器Selector上，注册SelectionKey.OP_READ；如果没有直接连接成功，则说明服务端没有返回TCP握手应答消息，但这并不代表连接失败。需要将SocketChannel注册到多路复用器Selector上，注册SelectionKey.OP_CONNECT，当服务端返回TCP syn-ack消息后，Selector就能够轮询到这个SocketChannel处于连接就绪状态。

        3、run()方法中while循环，轮询多路复用器Selector。当有就绪的Channel时，执行handleInput(key)方法。

        4、handleInput(key)方法：首先对SelectionKey进行判断，看它处于什么状态。如果处于连接状态，说明服务端已经返回ACK应答消息。这是需要对连接结果进行判断，调用SocketChannel的finishConnect方法。如果返回值为true，说明客户端连接成功；如果返回值为false或者抛出异常，说明连接失败。本例中，返回值为true，说明连接成功。将SocketChannel注册到多路复用器上，注册SelectionKey.OP_READ操作位，监听网络读操作，然后发送请求消息给服务端。

        5、doWrite方法：构造请求消息体，然后对其编码，写入到发送缓冲区中，最后调用SocketChannel的write方法进行发送。由于发送是异步的，所以会存在“半包写”问题，此处不再赘述。最后通过hasRemaining()方法对发送结果进行判断，如果缓冲区中的消息全部发送完成，打印“发送 2 服务成功”。

7 使用NIO编程优点

        1、客户端发起的连接操作是异步的，可以通过在多路复用器注册OP_CONNECT等待后续结果，不需要像之前那样被同步阻塞。

        2、SocketChannel的读写操作都是异步的，如果没有可读写的数据它不会同步等待，直接返回。这样I/O通信线程就可以处理其他的链路，不需要同步等待这个链路可用。

        3、线程模型优化：由于JDK的Selector在Linux等主流操作系统上通过epoll实现，它没有连接句柄数的限制，这意味一个Selector线程可以同时处理成千上万个客户端连接，而且性能不会随着客户端的增加而线性下降。因此，非常适合做高性能、高负载的网络服务器。

       

        

原文地址：https://blog.csdn.net/geminigoth/article/details/136295978

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！