Netty条记3：NIO编程

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程序：
必要注意的是：

• 举行网络传输时，涉及到的数据，必须要颠末缓冲区，不管是发送还是接收，团结用户态和内核态的切换过程就可以明白；
  
• NIO中的缓冲可以利用堆内存缓存和直接内存缓冲，这个必要团结零拷贝技术可以明白；
  
• 多路复用利用selector模式，必要循环遍历socket；
  

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

1. public class ServerHandle implements Runnable{
2.     private Selector selector;
3.     private ServerSocketChannel serverSocketChannel;
5.     public ServerHandle(int port) {
6.         try {
7.             selector = Selector.open();
8.             serverSocketChannel = ServerSocketChannel.open();
9.             // channel必须处于非阻塞模式下，不然会报错，所以不能同FileChannel一起使用
10.             serverSocketChannel.configureBlocking(false);
11.             serverSocketChannel.socket().bind(new InetSocketAddress(port));
12.             // 注册对应的事件，这里注册的是accept事件，只要监听到就会调用对应的处理器
13.             // 注册还可以添加附加对象，也就是第三个参数，获取方式：key.attachment();
14.             // 如果注册了事件，需要取消，需要调用channel.cancel()
15.             serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
16.             System.out.println("服务端已准备好：" + port);
17.         } catch (IOException e) {
18.             throw new RuntimeException(e);
19.         }
20.     }
22.     @Override
23.     public void run() {
24.         while (true) {
25.             try {
26.                 // 阻塞直到通道就绪，这边设置了超时时间
27.                 // 返回值：有多少通道就绪
28.                 selector.select(1000);
29.                 // 在通道就绪时，获取对应的键，也就是事件
30.                 Set<SelectionKey> selectionKeys = selector.selectedKeys();
31.                 Iterator<SelectionKey> iterator = selectionKeys.iterator();
32.                 while (iterator.hasNext()) {
33.                     SelectionKey key = iterator.next();
34.                     iterator.remove();
35.                     // 根据事件的类型进行对应的处理
36.                     handlerInput(key);
37.                 }
38.             } catch (Exception e) {
39.                 throw new RuntimeException(e);
40.             }
41.         }
42.     }
44.     private void handlerInput(SelectionKey key) throws IOException {
45.         // 在这个循环中，可能存在key.cancel() 或者移除，所以这里需要判断是否有效
46.         if (!key.isValid()) {
47.             return;
48.         }
49.         try {
50.             if (key.isAcceptable()) {
51.                 // 这里处理客户端连接服务端的事件
52.                 ServerSocketChannel channel = (ServerSocketChannel)key.channel();
53.                 try {
54.                     // 接收请求
55.                     SocketChannel sc = channel.accept();
56.                     System.out.println("-----建立连接------");
57.                     // 设置该通道非阻塞
58.                     sc.configureBlocking(false);
59.                     // 并注册read事件，监听着
60.                     sc.register(selector, SelectionKey.OP_READ);
61.                 } catch (IOException e) {
62.                     System.out.println("连接客户端失败！");
63.                     key.cancel();
64.                     channel.close();
65.                 }
66.             }
67.             if (key.isReadable()) {
68.                 SocketChannel sc = (SocketChannel)key.channel();
69.                 ByteBuffer buffer = ByteBuffer.allocate(1024);
70.                 int read = sc.read(buffer);
71.                 if (read > 0) {
72.                     // 反转：将这个缓冲中的数据从现在的位置变成从0开始
73.                     buffer.flip();
74.                     byte[] bytes = new byte[buffer.remaining()];
75.                     buffer.get(bytes);
76.                     String msg = new String(bytes, StandardCharsets.UTF_8);
77.                     System.out.println("服务器收到消息：" + msg);
79.                     doWrite(sc, "hello，收到消息：" + msg);
80.                 }
81.             }
82.         } catch (IOException e) {
83.             System.out.println("数据处理失败！");
84.             // 再处理失败，或异常时要退出通道，不然每次循环都会检查通道导致一致报错
85.             key.channel().close();
86.             key.cancel();
87.         }
88.     }
90.     private void doWrite(SocketChannel sc, String msg) throws IOException {
91.         byte[] bytes = msg.getBytes(StandardCharsets.UTF_8);
92.         ByteBuffer buffer = ByteBuffer.allocate(bytes.length);
93.         buffer.put(bytes);
94.         buffer.flip();
95.         sc.write(buffer);
97.     }
98. }

复制代码

服务端启动：

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

复制代码

客户端：

1. public class ClientHandle implements Runnable{
3.     private String ip;
4.     private int port;
5.     private SocketChannel socketChannel;
6.     private Selector selector;
8.     public ClientHandle(String ip, int port) {
9.         try {
10.             this.ip = ip;
11.             this.port = port;
12.             selector = Selector.open();
13.             socketChannel = SocketChannel.open();
14.             // 非阻塞状态
15.             socketChannel.configureBlocking(false);
16.         } catch (IOException e) {
17.             throw new RuntimeException(e);
18.         }
19.     }
21.     @Override
22.     public void run() {
23.         try {
24.             // 启动后，执行连接操作
25.             if (socketChannel.connect(new InetSocketAddress(ip, port))) {
26.                 // 连接服务端成功后，注册读取事件
27.                 socketChannel.register(selector, SelectionKey.OP_READ);
28.             } else {
29.                 // 如果连接失败，则再注册连接事件，之后再进行处理
30.                 socketChannel.register(selector, SelectionKey.OP_CONNECT);
31.             }
32.         } catch (IOException e) {
33.             throw new RuntimeException(e);
34.         }
36.         while (true) {
37.             try {
38.                 // 阻塞1000秒直到有通道就绪
39.                 selector.select(1000);
40.                 Set<SelectionKey> selectionKeys = selector.selectedKeys();
41.                 Iterator<SelectionKey> iterator = selectionKeys.iterator();
42.                 while (iterator.hasNext()) {
43.                     SelectionKey key = iterator.next();
44.                     iterator.remove();
45.                     handleInput(key);
46.                 }
47.             } catch (IOException e) {
48.                 throw new RuntimeException(e);
49.             }
50.         }
51.     }
53.     public void sendMsg(String msg) throws IOException {
54.         byte[] bytes = msg.getBytes(StandardCharsets.UTF_8);
55.         ByteBuffer buffer = ByteBuffer.allocate(bytes.length);
56.         buffer.put(bytes);
57.         buffer.flip();
58.         socketChannel.write(buffer);
59.     }
61.     private void handleInput(SelectionKey key) throws IOException {
62.         if (!key.isValid()) {
63.             return;
64.         }
65.         SocketChannel sc = (SocketChannel) key.channel();
66.         if (key.isConnectable()) {
67.             if (sc.finishConnect()) {
68.                 socketChannel.register(selector, SelectionKey.OP_READ);
69.             } else {
70.                 System.exit(1);
71.             }
72.         }
73.         if (key.isReadable()) {
74.             ByteBuffer buffer = ByteBuffer.allocate(1024);
75.             int read = sc.read(buffer);
76.             if (read > 0) {
77.                 buffer.flip();
78.                 byte[] bytes = new byte[buffer.remaining()];
79.                 buffer.get(bytes);
80.                 String msg = new String(bytes, StandardCharsets.UTF_8);
81.                 System.out.println("客户端收到消息：" + msg);
82.             } else if (read < 0) {
83.                 key.cancel();;
84.                 sc.close();
85.             }
86.         }
87.     }
88. }

复制代码

客户端启动：

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

复制代码

总结

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

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。