Netty笔记1：线程模型

Netty笔记2：零拷贝

Netty笔记3：NIO编程

Netty笔记4：Epoll

Netty笔记5：Netty开发实例

Netty笔记6：Netty组件

Netty笔记7：ChannelPromise通知处理

Netty笔记8：ByteBuf使用介绍

Netty笔记9：粘包半包

Netty笔记10：LengthFieldBasedFrameDecoder很简单

Netty笔记11：编解码器

Netty笔记12：模拟Web服务器

Netty笔记13：序列化

前言

想要快速理解NIO编程，需要先理解上篇的零拷贝技术和线程模型，本篇是对这两个知识的实践，也是netty的过度。

编程示例

我们尝试写一个NIO程序：

需要注意的是：

1. 进行网络传输时，涉及到的数据，必须要经过缓冲区，不管是发送还是接收，结合用户态和内核态的切换过程就可以明白；
2. NIO中的缓冲可以使用堆内存缓存和直接内存缓冲，这个需要结合零拷贝技术可以理解；
3. 多路复用使用selector模式，需要循环遍历socket；

注：buf在堆上。在进行数据发送时，如果使用堆内存，在JVM之外创建一个DirectBuf，然后把堆上的数据拷贝的这个DirectBuf，再写到SendBuf中，因为JVM中存在GC机制，如果使用引用方式，在拷贝过程中出现GC，会重新分配地址，导致数据出现问题。

服务端：

public class ServerHandle implements Runnable{private Selector selector;private ServerSocketChannel serverSocketChannel;public ServerHandle(int port) {try {selector = Selector.open();serverSocketChannel = ServerSocketChannel.open();// channel必须处于非阻塞模式下，不然会报错，所以不能同FileChannel一起使用serverSocketChannel.configureBlocking(false);serverSocketChannel.socket().bind(new InetSocketAddress(port));// 注册对应的事件，这里注册的是accept事件，只要监听到就会调用对应的处理器// 注册还可以添加附加对象，也就是第三个参数，获取方式：key.attachment();// 如果注册了事件，需要取消，需要调用channel.cancel()serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);System.out.println("服务端已准备好：" + port);} catch (IOException e) {throw new RuntimeException(e);}}@Overridepublic void run() {while (true) {try {// 阻塞直到通道就绪，这边设置了超时时间// 返回值：有多少通道就绪selector.select(1000);// 在通道就绪时，获取对应的键，也就是事件Set<SelectionKey> selectionKeys = selector.selectedKeys();Iterator<SelectionKey> iterator = selectionKeys.iterator();while (iterator.hasNext()) {SelectionKey key = iterator.next();iterator.remove();// 根据事件的类型进行对应的处理handlerInput(key);}} catch (Exception e) {throw new RuntimeException(e);}}}private void handlerInput(SelectionKey key) throws IOException {// 在这个循环中，可能存在key.cancel() 或者移除，所以这里需要判断是否有效if (!key.isValid()) {return;}try {if (key.isAcceptable()) {// 这里处理客户端连接服务端的事件ServerSocketChannel channel = (ServerSocketChannel)key.channel();try {// 接收请求SocketChannel sc = channel.accept();System.out.println("-----建立连接------");// 设置该通道非阻塞sc.configureBlocking(false);// 并注册read事件，监听着sc.register(selector, SelectionKey.OP_READ);} catch (IOException e) {System.out.println("连接客户端失败！");key.cancel();channel.close();}}if (key.isReadable()) {SocketChannel sc = (SocketChannel)key.channel();ByteBuffer buffer = ByteBuffer.allocate(1024);int read = sc.read(buffer);if (read > 0) {// 反转：将这个缓冲中的数据从现在的位置变成从0开始buffer.flip();byte[] bytes = new byte[buffer.remaining()];buffer.get(bytes);String msg = new String(bytes, StandardCharsets.UTF_8);System.out.println("服务器收到消息：" + msg);doWrite(sc, "hello，收到消息：" + msg);}}} catch (IOException e) {System.out.println("数据处理失败！");// 再处理失败，或异常时要退出通道，不然每次循环都会检查通道导致一致报错key.channel().close();key.cancel();}}private void doWrite(SocketChannel sc, String msg) throws IOException {byte[] bytes = msg.getBytes(StandardCharsets.UTF_8);ByteBuffer buffer = ByteBuffer.allocate(bytes.length);buffer.put(bytes);buffer.flip();sc.write(buffer);}
}

服务端启动：

    public static void main(String[] args) {ServerHandle serverHandle = new ServerHandle(8080);new Thread(serverHandle).start();}

客户端：

public class ClientHandle implements Runnable{private String ip;private int port;private SocketChannel socketChannel;private Selector selector;public ClientHandle(String ip, int port) {try {this.ip = ip;this.port = port;selector = Selector.open();socketChannel = SocketChannel.open();// 非阻塞状态socketChannel.configureBlocking(false);} catch (IOException e) {throw new RuntimeException(e);}}@Overridepublic void run() {try {// 启动后，执行连接操作if (socketChannel.connect(new InetSocketAddress(ip, port))) {// 连接服务端成功后，注册读取事件socketChannel.register(selector, SelectionKey.OP_READ);} else {// 如果连接失败，则再注册连接事件，之后再进行处理socketChannel.register(selector, SelectionKey.OP_CONNECT);}} catch (IOException e) {throw new RuntimeException(e);}while (true) {try {// 阻塞1000秒直到有通道就绪selector.select(1000);Set<SelectionKey> selectionKeys = selector.selectedKeys();Iterator<SelectionKey> iterator = selectionKeys.iterator();while (iterator.hasNext()) {SelectionKey key = iterator.next();iterator.remove();handleInput(key);}} catch (IOException e) {throw new RuntimeException(e);}}}public void sendMsg(String msg) throws IOException {byte[] bytes = msg.getBytes(StandardCharsets.UTF_8);ByteBuffer buffer = ByteBuffer.allocate(bytes.length);buffer.put(bytes);buffer.flip();socketChannel.write(buffer);}private void handleInput(SelectionKey key) throws IOException {if (!key.isValid()) {return;}SocketChannel sc = (SocketChannel) key.channel();if (key.isConnectable()) {if (sc.finishConnect()) {socketChannel.register(selector, SelectionKey.OP_READ);} else {System.exit(1);}}if (key.isReadable()) {ByteBuffer buffer = ByteBuffer.allocate(1024);int read = sc.read(buffer);if (read > 0) {buffer.flip();byte[] bytes = new byte[buffer.remaining()];buffer.get(bytes);String msg = new String(bytes, StandardCharsets.UTF_8);System.out.println("客户端收到消息：" + msg);} else if (read < 0) {key.cancel();;sc.close();}}}
}

客户端启动：

 public static void main(String[] args) throws IOException {ClientHandle handle = new ClientHandle("localhost", 8080);new Thread(handle).start();Scanner scanner = new Scanner(System.in);// 死循环保持监听while (true) {// 每次控制台输入，就发送给服务端handle.sendMsg(scanner.nextLine());}}

总结

该示例对应于reactor单线程模型，服务端是一个单线程，通过selector单线程循环接收客户端的请求，并识别客户端请求事件类型，进行分发处理，相对于BIO很明显的区别，就是它不会等到上一个请求处理完成。在消息接收与发送的过程中，我们需要对缓冲数据进行处理，也就是对应于零拷贝知识点中提到的缓冲区概念，实例中用到了ByteBuffer对象，它是NIO中一个比较重要的对象，下一篇会进行说明。