| 版本 | 内容 | 时间 | | ---- | ---- | --------------------- | | V1 | 新建 | 2021年9月22日07:56:54 | # NIO网络编程案例 ## 处理accept事件客户端 ```java public class Client { public static void main(String[] args) { try (SocketChannel socketChannel = SocketChannel.open()) { socketChannel.configureBlocking(false); socketChannel.connect(new InetSocketAddress(10086)); ByteBuffer buffer = ByteBuffer.allocate(16).put("憨八龟".getBytes()); socketChannel.write(buffer); } catch (IOException e) { e.printStackTrace(); } } } ``` 服务端： ```java @Slf4j public class Server { public static void main(String[] args) { try (ServerSocketChannel channel = ServerSocketChannel.open()) { // 设置非阻塞 channel.configureBlocking(false); // 绑定端口 channel.bind(new InetSocketAddress(10086)); // 获取选择器对象 Selector selector = Selector.open(); // 将通道注册到选择器上，设置感兴趣的事件为 accept // accept 服务端成功接收连接时触发 channel.register(selector, SelectionKey.OP_ACCEPT); while (true) { int count = selector.select(); log.debug("有 {} 个感兴趣的事件", count); if (count <= 0) { continue; } // 获取已发生的事件 Iterator iterator = selector.selectedKeys().iterator(); // 遍历事件进行处理 while (iterator.hasNext()) { SelectionKey key = iterator.next(); if (key.isAcceptable()) { ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel(); // 获取客户端的SocketChannel SocketChannel socketChannel = serverSocketChannel.accept(); // 设置非阻塞 socketChannel.configureBlocking(false); log.debug("获得了一个客户端的连接: {}", socketChannel); // 给得到的客户端的clientChannel注册到Selector上 // 给每个添加一个自己的Buffer ByteBuffer privateBuffer = ByteBuffer.allocate(16); // 设置事件为可读，数据可读入时触发，因为是客户端发起的连接请求事件, socketChannel.register(selector, SelectionKey.OP_READ, privateBuffer); log.debug("客户端Channel绑定选择器成功: {}", socketChannel); } else if (key.isReadable()) { log.debug("客户端Channel可读..,取消读取"); // 本案例不考虑读操作，取消掉 key.cancel(); } // 移除已经处理的事件 iterator.remove(); } } } catch (Exception e) { e.printStackTrace(); } } } ``` ## 处理read事件 ### 案例客户端： ```java /** * 测试read事件 */ public class Client { public static void main(String[] args) { try (SocketChannel socketChannel = SocketChannel.open()) { // socketChannel.configureBlocking(false); socketChannel.connect(new InetSocketAddress(10086)); ByteBuffer buffer = Charset.defaultCharset().encode("hello"); System.out.println("..."); socketChannel.write(buffer); } catch (IOException e) { e.printStackTrace(); } } } ``` 服务端： ```java @Slf4j public class Server { public static void main(String[] args) { try { ServerSocketChannel channel = ServerSocketChannel.open(); channel.configureBlocking(false); channel.bind(new InetSocketAddress(10086)); Selector selector = Selector.open(); channel.register(selector, SelectionKey.OP_ACCEPT); while (true) { int count = selector.select(); if (count <= 0) { continue; } Iterator iterator = selector.selectedKeys().iterator(); while (iterator.hasNext()) { SelectionKey key = iterator.next(); if (key.isAcceptable()) { // 有客户端连接 ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel(); SocketChannel socketChannel = serverChannel.accept(); socketChannel.configureBlocking(false); // 客户端通道注册可读事件 ByteBuffer privateBuffer = ByteBuffer.allocate(16); socketChannel.register(selector, SelectionKey.OP_READ, privateBuffer); log.debug("客户端已连接: " + socketChannel); } else if (key.isReadable()) { try { // channel有可读事件 SocketChannel socketChannel = (SocketChannel) key.channel(); ByteBuffer buffer = (ByteBuffer) key.attachment(); int read = socketChannel.read(buffer); // 如果是客户端正常断开，read方法会返回-1 if (read == -1) { key.cancel(); socketChannel.close(); } else { buffer.flip(); String data = Charset.defaultCharset().decode(buffer).toString(); log.debug("接收到客户端的数据: " + data); buffer.clear(); } } catch (IOException e) { // 客户端异常断开，也会产生一个读事件，比如说直接停止程序了，此时会抛异常 e.printStackTrace(); // 取消客户端异常断开的读事件 key.cancel(); } } iterator.remove(); } } } catch (IOException e) { e.printStackTrace(); } } } ``` ### 消息边界和Buffer大小假如客户端使用UTF-8发送中文“你好”，此时占用6个字节。假如服务端使用的Buffer只有4个字节大小，那么此时就会乱码。 ``` 08:36:18 [DEBUG] [main] c.g.n.n.r.Server - 客户端已连接: java.nio.channels.SocketChannel[connected local=/192.168.3.11:10086 remote=/192.168.3.11:51101] 08:36:18 [DEBUG] [main] c.g.n.n.r.Server - 接收到客户端的数据: 你� 08:36:18 [DEBUG] [main] c.g.n.n.r.Server - 接收到客户端的数据: �� ``` 那么此时如何去处理消息边界呢。有如下几种思路吧。下面几种思路的实现，在Netty中都有现成的类。 1. **固定消息长度**。也就是说每次发送的数据的大小一致，服务器就按照固定长度去读取数据。 2. **按照分隔符去读取**。也就是约定一个指定的“字符”代表分隔符。 3. **TLV 格式，即 Type 类型、Length 长度、Value 数据，类型和长度已知的情况下，就可以方便获取消息大小**，分配合适的 buffer，缺点是 buffer 需要提前分配，如果内容过大，则影响 server 吞吐量。 * Http 1.1 是 TLV 格式 * Http 2.0 是 LTV 格式每个通道Channel的Buffer的大小分配需要做成可以**自动扩容和自动缩小**，Netty就是这样实现的。 ByteBuffer 不能太大，比如一个 ByteBuffer 1Mb 的话，要支持百万连接就要 1Tb 内存，首先分配一个较小的 buffer，例如 4k，如果发现数据不够，再分配 8k 的 buffer，将 4k buffer 内容拷贝至 8k buffer，优点是消息连续容易处理，缺点是数据拷贝耗费性能 ## 处理write事件在非阻塞模式下。可能无法一下子把buffer的数据写到Channel通道中，所以需要根据write方法的返回值来判断是否已经写完。客户端 ```java public class Client { public static void main(String[] args) throws IOException { SocketChannel sc = SocketChannel.open(); sc.connect(new InetSocketAddress(10086)); int count = 0; while (true) { ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024); count += sc.read(buffer); System.out.println(count); buffer.clear(); } } } ``` ### 有问题的write 在客户端连接到服务端时，向客户端发送长30000000的字符串。 ```java @Slf4j public class ServerProblem { public static void main(String[] args) { try (ServerSocketChannel ssc = ServerSocketChannel.open()) { ssc.configureBlocking(false); ssc.bind(new InetSocketAddress(10086)); Selector selector = Selector.open(); ssc.register(selector, SelectionKey.OP_ACCEPT); while (true) { int count = selector.select(); if (count <= 0) { continue; } Iterator iterator = selector.selectedKeys().iterator(); while (iterator.hasNext()) { SelectionKey key = iterator.next(); if (key.isAcceptable()) { ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel(); SocketChannel sc = serverChannel.accept(); sc.configureBlocking(false); sc.register(selector, SelectionKey.OP_READ); // 连接成功后，往客户端发送大量数据 StringBuilder sb = new StringBuilder(); for (int i = 0; i < 30000000; i++) { sb.append("a"); } ByteBuffer buffer = StandardCharsets.UTF_8.encode(sb.toString()); /* ---------- 问题处开始 --------*/ // 问题原因：假如此次写入到channel一次写不完，网络发送能力是有限的 // 所以可能在某些时刻，channel写满了，就写不进去了， // 而此时又在while循环里，导致线程一直卡在这个SelectionKey里 // 不满足nio的非阻塞的思想 while (buffer.hasRemaining()) { // 数据发送到Channel中 int write = sc.write(buffer); log.debug("此次一共写入数据: " + write); } /* ---------- 问题处结束 --------*/ } iterator.remove(); } } } catch (Exception e) { e.printStackTrace(); } } } ``` 现象：客户端完整接收数据，但是客户端有问题。客户端的部分打印日志： ``` 08:06:08 [DEBUG] [main] c.g.n.n.w.ServerProblem - 此次一共写入数据: 786426 08:06:08 [DEBUG] [main] c.g.n.n.w.ServerProblem - 此次一共写入数据: 1834994 08:06:08 [DEBUG] [main] c.g.n.n.w.ServerProblem - 此次一共写入数据: 0 08:06:08 [DEBUG] [main] c.g.n.n.w.ServerProblem - 此次一共写入数据: 0 08:06:08 [DEBUG] [main] c.g.n.n.w.ServerProblem - 此次一共写入数据: 786426 08:06:08 [DEBUG] [main] c.g.n.n.w.ServerProblem - 此次一共写入数据: 0 08:06:08 [DEBUG] [main] c.g.n.n.w.ServerProblem - 此次一共写入数据: 0 08:06:08 [DEBUG] [main] c.g.n.n.w.ServerProblem - 此次一共写入数据: 0 08:06:08 [DEBUG] [main] c.g.n.n.w.ServerProblem - 此次一共写入数据: 1834994 08:06:08 [DEBUG] [main] c.g.n.n.w.ServerProblem - 此次一共写入数据: 0 08:06:08 [DEBUG] [main] c.g.n.n.w.ServerProblem - 此次一共写入数据: 0 ``` 可以看到在while循环中，有些写入0数据的。这是因为无法一下子把buffer的数据全部写到Channel通道中，在某次循环中，Channel无法写入，所以有些循环中是写入0数据。这样在某些循环中无法写入数据，但是又卡在while循环中，不符合NIO的非阻塞的思想，所以需要改进。 ### 改进后的案例改进思路： - 在客户端连接到服务端时，先尝试发送数据，假如没一次发送完，就关注写事件。 - Selector 检查 Channel 上的可写事件，如果所有的数据写完了，就取消 Channel 的注册。假如不取消，会每次可写均会触发 write 事件 ```java @Slf4j public class Server { public static void main(String[] args) { try (ServerSocketChannel ssc = ServerSocketChannel.open()) { ssc.configureBlocking(false); ssc.bind(new InetSocketAddress(10086)); Selector selector = Selector.open(); ssc.register(selector, SelectionKey.OP_ACCEPT); while (true) { int count = selector.select(); if (count <= 0) { continue; } Iterator iterator = selector.selectedKeys().iterator(); while (iterator.hasNext()) { SelectionKey key = iterator.next(); if (key.isAcceptable()) { log.debug("client connecting..."); ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel(); SocketChannel sc = serverChannel.accept(); sc.configureBlocking(false); SelectionKey sKey = sc.register(selector, SelectionKey.OP_READ); // 连接成功后，往客户端发送大量数据 StringBuilder sb = new StringBuilder(); for (int i = 0; i < 30000000; i++) { sb.append("a"); } ByteBuffer buffer = StandardCharsets.UTF_8.encode(sb.toString()); // 数据发送到Channel中 int write = sc.write(buffer); log.debug("begin connect write data count: " + write); if (buffer.hasRemaining()) { // 假如此时Buffer中的数据未读完,注册一个可读事件 log.debug("ops = {}", sKey.interestOps()); sKey.interestOps(sKey.interestOps() + SelectionKey.OP_WRITE); // 将未读完的数据绑定 sKey.attach(buffer); } } else if (key.isWritable()) { log.debug("channel is writeable..."); SocketChannel sc = (SocketChannel) key.channel(); ByteBuffer buffer = (ByteBuffer) key.attachment(); int write = sc.write(buffer); log.debug("{} byte data was written this time", write); if (!buffer.hasRemaining()) { log.debug("write over, cancel listening Writable event..."); // 数据已经读取完了，取消关注可读事件，清空buffer key.interestOps(key.interestOps() - SelectionKey.OP_WRITE); key.attach(null); } } iterator.remove(); } } } catch (Exception e) { e.printStackTrace(); } } } ```

NIO网络编程案例.

Dylan Kwok

NIO网络编程案例