netty——IO、NIO、AIO进化之路

本文会说明各种IO的特点、分别解决了什么样的问题做一个分析阐述，并结合Java代码例子来辅助理解，像这些的历史演进和详细的底层原理网上很多，所以我们只站在应用层，使用者的角度去分析

（所有例子均可直接运行）

BIO——同步阻塞IO

看这个名称大家可能会有点陌生，我们直接上例子：

服务端：

public static void main(String[] args) throws IOException {//1.创建服务端Socket 并绑定端口ServerSocket serverSocket = new ServerSocket(8080);//2.等待客户端连接 阻塞的Socket accept = serverSocket.accept();System.out.println(accept.getRemoteSocketAddress() + " 客户端已连接");//3.获取输入、输出流InputStream inputStream = accept.getInputStream();OutputStream outputStream = accept.getOutputStream();//4.接收客户端信息byte[] bytes = new byte[1024];inputStream.read(bytes);String data = new String(bytes);System.out.println("来自" + accept.getRemoteSocketAddress() + "的信息：" + data);//5.返回信息outputStream.write(data.getBytes());accept.shutdownOutput();//6.关闭资源inputStream.close();outputStream.close();accept.close();serverSocket.close();}

客户端：

public static void main(String[] args) throws IOException {//1.创建客户端SocketSocket socket = new Socket("127.0.0.1",8080);//2.获取输入、输出流InputStream inputStream = socket.getInputStream();OutputStream outputStream = socket.getOutputStream();//3.给服务端发送信息outputStream.write("你好".getBytes());socket.shutdownOutput();//4.获取服务端返回信息byte[] data = new byte[1024];inputStream.read(data);System.out.println("来自服务端的信息：" + new String(data));//6.关闭资源inputStream.close();outputStream.close();socket.close();}

这就是我们熟知的Socket连接，也是Java最早的网络通信IO，为什么这种叫同步阻塞IO：

因为在做read操作、accept操作的时候会阻塞没法往下执行，说白了就是串行的，就因为这个服务端和客户端只能1对1通信，这合理嘛？肯定不合理啊，所以进阶的有了伪异步IO

伪异步阻塞IO

看完上面的，很多人就有想法了，你说同步的只能1对1通信，那我直接把服务端改成多线程版本不就好了嘛，不就可以1对多通信了嘛，没错这版本确实是这样，如下：

服务端：

public static void main(String[] args) throws IOException {//1.创建服务端Socket 并绑定端口ServerSocket serverSocket = new ServerSocket(8080);//2.等待客户端连接 多线程模式 (开线程异步等待)new Thread(()->{while (true){try {Socket accept = serverSocket.accept();System.out.println(accept.getRemoteSocketAddress() + " 客户端已连接");// 开线程异步处理客户端连接任务new Thread(new AcceptHandler(accept)).start();} catch (IOException e) {e.printStackTrace();}}}).start();// 阻塞防止程序退出while (true){}}private static class AcceptHandler implements Runnable{private Socket accept;private InputStream inputStream = null;private OutputStream outputStream =null;public AcceptHandler(Socket accept){this.accept=accept;}@Overridepublic void run() {try {//3.获取输入、输出流inputStream = accept.getInputStream();outputStream = accept.getOutputStream();//4.接收客户端信息byte[] bytes = new byte[1024];inputStream.read(bytes);String data = new String(bytes);if(data!=null){System.out.println("来自" + accept.getRemoteSocketAddress() + "的信息：" + data);//5.返回信息outputStream.write(data.getBytes());accept.shutdownOutput();}} catch (IOException e) {System.out.println(accept.getRemoteSocketAddress() + "发送异常断开连接");closeSource();}finally {System.out.println(accept.getRemoteSocketAddress() + "断开连接");closeSource();}}private void closeSource(){//6.关闭资源try {if(inputStream!=null){inputStream.close();}if(outputStream!=null){outputStream.close();}accept.close();} catch (IOException ioException) {ioException.printStackTrace();}}}

客户端不变，服务端我们做了三个改动：

• 一：在等待客户端连接的时候我们开启一个线程，并死循环等待连接，这样可以保证不阻塞主线程的运行，同时可以不断的和客户端建立连接
• 二：和客户端建立连接后又开启一个线程来单独处理与客户端的通信
• 三：最后加了个死循环防止程序退出，因为现在是异步的了

这样处理不就是异步的了吗？为什么叫伪异步阻塞IO呢？

虽然现在不会阻塞主线程了，但是阻塞并没有解决，该阻塞的地方依旧还是会阻塞，所以本质上来说只是解决了1对1连接通信的问题

但是新的问题又来了，现在虽然是1对多通信，但是有一个客户端连接就新建一个线程，1万个客户端就1万个线程，这合理吗？这明显不合理啊，用线程池管理？那也不行啊，这连接一多还要排队吗？极端情况下，队列不一样会爆？

那怎么办？有没有可能一个线程监听多个连接呢？于是有了NIO

NIO——同步非阻塞IO

NIO的引入同时引入了三个概念ByteBuffer缓冲区、Channel通道和Selector多路复用器

• Channel的作用：就是一个通道，数据读取和写入的通道，根据功能可以分为不同的通道如：网络通道ServerSocketChannel和SocketChannel、文件操作通道FileChannel等等
• Selector的作用：是轮询Channel上面的事件，如读事件、写事件、连接事件、接受连接事件
• ByteBuffer缓冲区：就是向Channel读取或写入数据的对象，本质就是个字节数组

怎么理解这三个呢？说白了以传统IO为例：服务端accept就是接受连接事件、客户端connect就是连接事件、发送消息就是写事件、读取消息就是读事件
Selector就是监听这些事件的工具
ServerSocketChannel是服务端接受连接的通道，所以只能注册监听连接事件
SocketChannel是服务端与客户端连接建立后的通道，所以可以注册读写事件、连接事件
ByteBuffer就是Channel读取或写入数据的单位对象

下面搞个例子看看，注释全有：

服务端：

public static void main(String[] args) throws IOException {// 开启服务端Socket通道ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();// 设置为非阻塞serverSocketChannel.configureBlocking(false);// 绑定端口serverSocketChannel.socket().bind(new InetSocketAddress(8080));// 打开多路复用器 并将其注册到通道上 监听连接请求事件Selector selector = Selector.open();// 为服务端Socket通道 注册一个接受连接的事件 // 假设有客户端要连接 下面轮询的时候就会触发这个事件 我们就可以去与客户端建立连接了serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);while (true) {// 这段时间没获取到任何事件，则跳过下面操作// 不同于IO和BIO的阻塞 多路复用器会一直轮询 如果长时间无事件 这里会一直空循环// 所以这里在查询事件的时候加了个时间 这样无事件的情况下 1s才会循环一次if (selector.select(1000) == 0) {continue;}// 获取到本次轮询所获取到的全部事件Iterator<SelectionKey> selectorKeys = selector.selectedKeys().iterator();// 轮询获取到的事件，并处理while (selectorKeys.hasNext()) {SelectionKey selectorKey = selectorKeys.next();//这个已经处理的事件Key一定要移除。如果不移除，就会一直存在在selector.selectedKeys集合中//待到下一次selector.select() > 0时，这个Key又会被处理一次selectorKeys.remove();try {// 事件key处理 也就是事件处理selectorKeyHandler(selectorKey, selector);} catch (Exception e) {SocketChannel channel = (SocketChannel) selectorKey.channel();System.out.println(channel.getRemoteAddress() + "客户端已断开连接");if (selectorKey != null) {selectorKey.cancel();if (selectorKey.channel() != null) {selectorKey.channel().close();}}}}}}// 事件处理方法 按照事件类型处理不同的事件public static void selectorKeyHandler(SelectionKey selectorKey, Selector selector) throws IOException {// 连接事件 代表有客户端连接 所以需要去处理这个连接请求if (selectorKey.isAcceptable()) {acceptHandler(selectorKey, selector);}// 读事件 可以去读取信息if (selectorKey.isReadable()) {readHandler(selectorKey, selector);}// 写事件 可以向客户端发送信息if (selectorKey.isWritable()) {SocketChannel socketChannel = (SocketChannel) selectorKey.channel();writeHandler(socketChannel);// 写事件完成后要取消写事件不然会一直写  我这里就干脆注册了个读事件socketChannel.register(selector,SelectionKey.OP_READ);}}// 连接事件处理 这个有客户端要建立连接了  所以accept与客户端建立连接public static void acceptHandler(SelectionKey selectorKey, Selector selector) throws IOException {ServerSocketChannel channel = (ServerSocketChannel) selectorKey.channel();SocketChannel accept = channel.accept();// 建立连接后 客户端和服务端就等于形成了一个数据交互的通道 SocketChannel// 这个通道也要设置为非阻塞accept.configureBlocking(false);// 为这个通道注册一个读事件 表示我先读取客户端信息accept.register(selector, SelectionKey.OP_READ);System.out.println(accept.getRemoteAddress() + "客户端已连接");}// 读事件处理  读取客户端的信息public static void readHandler(SelectionKey selectorKey, Selector selector) throws IOException {SocketChannel channel = (SocketChannel) selectorKey.channel();ByteBuffer allocate = ByteBuffer.allocate(1024);int read = channel.read(allocate);if (read > 0) {allocate.flip();byte[] bytes = new byte[allocate.remaining()];allocate.get(bytes);System.out.println(channel.getRemoteAddress() + "发来消息：" + new String(bytes));}if(read<0){System.out.println(channel.getRemoteAddress() + "断开连接");}// 读完信息后要给客户端发送信息 所以这个再注册一个写的事件channel.register(selector, SelectionKey.OP_WRITE);}// 写事件处理public static void writeHandler(SocketChannel socketChannel) throws IOException {byte[] bytes = "你好".getBytes();ByteBuffer allocate = ByteBuffer.allocate(bytes.length);allocate.put(bytes);allocate.flip();socketChannel.write(allocate);}

客户端：

public static void main(String[] args) throws IOException {// 开启一个Socket通道SocketChannel clientChannel = SocketChannel.open();// 设置非阻塞clientChannel.configureBlocking(false);// 允许端口复用clientChannel.socket().setReuseAddress(true);// 连接地址clientChannel.connect(new InetSocketAddress("127.0.0.1", 8080));// 开启多路复用器Selector selector = Selector.open();// 为这个通道注册一个连接事件clientChannel.register(selector, SelectionKey.OP_CONNECT);while (true) {// 这段时间没获取到任何事件，则跳过下面操作// 不同于IO和BIO的阻塞 多路复用器会一直轮询 如果长时间无事件 这里会一直空循环// 所以这里在查询事件的时候加了个时间 这样无事件的情况下 1s才会循环一次if (selector.select(1000) == 0) {continue;}// 获取到本次轮询所获取到的全部事件Iterator<SelectionKey> selectorKeys = selector.selectedKeys().iterator();// 轮询获取到的事件，并处理while (selectorKeys.hasNext()) {SelectionKey selectorKey = selectorKeys.next();//这个已经处理的事件Key一定要移除。如果不移除，就会一直存在在selector.selectedKeys集合中//待到下一次selector.select() > 0时，这个Key又会被处理一次selectorKeys.remove();try {// 事件key处理selectorKeyHandler(selectorKey, selector);} catch (Exception e) {if (selectorKey != null) {selectorKey.cancel();if (selectorKey.channel() != null) {selectorKey.channel().close();}}}}}}// 事件处理方法public static void selectorKeyHandler(SelectionKey selectorKey, Selector selector) throws IOException {// 连接事件 判断是否连接成功if (selectorKey.isValid()) {SocketChannel channel = (SocketChannel) selectorKey.channel();if (selectorKey.isConnectable() && channel.finishConnect()) {System.out.println("连接成功........");// 连接成功注册写事件 向服务端发送信息channel.register(selector,SelectionKey.OP_WRITE);}}// 读事件 可以去读取信息if (selectorKey.isReadable()) {readHandler(selectorKey, selector);}// 写事件 可以向客户端发送信息if (selectorKey.isWritable()) {SocketChannel channel = (SocketChannel) selectorKey.channel();writeHandler(channel);// 写事件完成后要取消写事件不然会一直写  我这里就干脆注册了个读事件channel.register(selector,SelectionKey.OP_READ);}}// 读事件处理  就是处理服务端发来的消息public static void readHandler(SelectionKey selectorKey, Selector selector) throws IOException {SocketChannel channel = (SocketChannel) selectorKey.channel();ByteBuffer allocate = ByteBuffer.allocate(1024);int read = channel.read(allocate);if (read > 0) {allocate.flip();byte[] bytes = new byte[allocate.remaining()];allocate.get(bytes);System.out.println("服务端发来消息：" + new String(bytes));}if(read<0){System.out.println("与服务端断开连接");}}// 写事件处理 就是像服务端发送消息public static void writeHandler(SocketChannel socketChannel) throws IOException {byte[] bytes = "你好".getBytes();ByteBuffer allocate = ByteBuffer.allocate(bytes.length);allocate.put(bytes);allocate.flip();socketChannel.write(allocate);}

可以看到写法和传统的IO完全不一样了，操作的对象都是Channel，读写对象都是ByteBuffer，那到底是什么引起了这种改变呢？因为系统内核的优化，说白了这种操作都是API，底层都是需要系统支持的，系统在这块也有一个模型优化，简单介绍三种模型区别：

• select： 每有一个连接的产生会打开一个Socket描述符(下面简称FD)，select会把这些FD保存在一个数组中，因为是数组所以就代表有了容量的上限意味了连接数量的上限，每次调用，都会遍历这个数组，1w个连接就算只有一个事件，也会遍历这1w个连接，效率极低
• poll： 和select不同，这个底层结构是链表，所有没了连接数量的上限，但是每次调用依旧会遍历所有的
• epoll： 底层结构是红黑树，同样没有连接数量的上限，而且有一个就绪的事件列表，这意味着不再需要遍历所有的连接了

JDK中采用的就是epoll模型，但尽管这样也依旧是同步的，因为还是需要主动去获取结果，只是从方式阻塞等待变成了轮询，有没有什么方式在结果产生的时候异步的回调呢？于是有了AIO

AIO——异步IO

这种方式同样需要系统的支持，目前主流还是NIO，这块就不多介绍了，提供个例子：

服务端：

    public static void main(String[] args) throws IOException {AsynchronousServerSocketChannel serverSocketChannel = AsynchronousServerSocketChannel.open();serverSocketChannel.bind(new InetSocketAddress(8080));// 接收连接的时候 提供连接处理类serverSocketChannel.accept(serverSocketChannel, new ServerSocketHandler());// 异步的  防止程序退出while (true) {}}// 连接处理public static class ServerSocketHandler implements CompletionHandler<AsynchronousSocketChannel, AsynchronousServerSocketChannel> {@Overridepublic void completed(AsynchronousSocketChannel result, AsynchronousServerSocketChannel attachment) {// 继续接受连接attachment.accept(attachment, this);try {System.out.println(result.getRemoteAddress() + " 已连接");} catch (IOException e) {e.printStackTrace();}new Thread(() -> {// 异步读readHandler(result);}).start();// 写数据处理writeHandler(result, "你好");}@Overridepublic void failed(Throwable exc, AsynchronousServerSocketChannel attachment) {System.out.println("发生异常");}public void readHandler(AsynchronousSocketChannel socketChannel) {ByteBuffer allocate = ByteBuffer.allocate(1024);socketChannel.read(allocate, allocate, new CompletionHandler<Integer, ByteBuffer>() {@Overridepublic void completed(Integer result, ByteBuffer attachment) {try {if (result > 0) {attachment.flip();byte[] bytes = new byte[attachment.remaining()];attachment.get(bytes);System.out.println(socketChannel.getRemoteAddress() + " 客户端消息: " + new String(bytes));readHandler(socketChannel);}} catch (IOException e) {e.printStackTrace();}}@Overridepublic void failed(Throwable exc, ByteBuffer attachment) {System.out.println();try {System.out.println(socketChannel.getRemoteAddress() + " 已下线");socketChannel.close();} catch (IOException e) {e.printStackTrace();}}});}public void writeHandler(AsynchronousSocketChannel socketChannel, String data) {byte[] bytes = data.getBytes();ByteBuffer allocate = ByteBuffer.allocate(bytes.length);allocate.put(bytes);allocate.flip();socketChannel.write(allocate, allocate, new CompletionHandler<Integer, ByteBuffer>() {@Overridepublic void completed(Integer result, ByteBuffer attachment) {if (attachment.hasRemaining()) {socketChannel.write(attachment, attachment, this);}}@Overridepublic void failed(Throwable exc, ByteBuffer attachment) {try {socketChannel.close();} catch (IOException e) {e.printStackTrace();}}});}}

客户端：

 public static void main(String[] args) throws IOException {AsynchronousSocketChannel socketChannel=AsynchronousSocketChannel.open();socketChannel.connect(new InetSocketAddress("127.0.0.1", 8080), null, new AsyncClientHandler(socketChannel));while (true){}}public static class AsyncClientHandler implements CompletionHandler<Void, AsyncClientHandler>{private AsynchronousSocketChannel socketChannel;public AsyncClientHandler(AsynchronousSocketChannel socketChannel){this.socketChannel=socketChannel;}@Overridepublic void completed(Void result, AsyncClientHandler attachment) {new Thread(()->{// 异步 一秒发送一次消息while (true){writeHandler("你好");try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}}}).start();// 读处理readHandler();}@Overridepublic void failed(Throwable exc, AsyncClientHandler attachment) {}public void readHandler() {ByteBuffer allocate = ByteBuffer.allocate(1024);socketChannel.read(allocate, allocate, new CompletionHandler<Integer, ByteBuffer>() {@Overridepublic void completed(Integer result, ByteBuffer attachment) {attachment.flip();byte[] bytes = new byte[attachment.remaining()];attachment.get(bytes);System.out.println(" 服务端消息: " + new String(bytes));}@Overridepublic void failed(Throwable exc, ByteBuffer attachment) {try {socketChannel.close();} catch (IOException e) {e.printStackTrace();}}});}public void writeHandler( String data) {byte[] bytes = data.getBytes();ByteBuffer allocate = ByteBuffer.allocate(bytes.length);allocate.put(bytes);allocate.flip();socketChannel.write(allocate, allocate, new CompletionHandler<Integer, ByteBuffer>() {@Overridepublic void completed(Integer result, ByteBuffer attachment) {if (attachment.hasRemaining()) {socketChannel.write(attachment, attachment, this);}}@Overridepublic void failed(Throwable exc, ByteBuffer attachment) {try {socketChannel.close();} catch (IOException e) {e.printStackTrace();}}});}}

总结

	BIO	伪异步IO	NIO	AIO
线程:客户端	1:1	N:M (M可以大于N)	1:N (一个线程处理多个)	0:M (无需额外线程,异步回调)
I/O类型	同步阻塞	伪异步阻塞	同步非阻塞	异步非阻塞
可靠性	非常差	差	高	高
难度	简单	简单	复杂	复杂
性能	低	中	高	高