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方法内轮询准备就绪的keyint 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;}@Overridepublic 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);}}@Overridepublic 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 服务成功”。