JAVA BIO,NIO,AIO详解（附代码实现）以及Netty的简介

布满荆棘的人生 2022-05-25 05:42 115阅读 0赞

# 缘起 #

NIO基本是面试过程中必问专题，很有了解的必要。

# Java中的三种IO模式 #

**BIO**：同步堵塞  
**NIO**：同步非堵塞IO，JDK1.4提出  
**AIO**：异步非堵塞，在JDK1.7中才被提出

在JAVA中，IO分两块，一块是操作文件的，一块是操作网络的。本文主要对操作网络的这一块进行说明

# 网络IO #

首先我们要明白的是，所谓Nio，Aio的提出，都只是为了加快服务器端的处理能力的，而非客户端。为了能够通俗的理解BIO,NIO,AIO，我们这里举个例子。

## 通俗的解释 ##

**参演人员**：客人（客户端），酒吧（服务端），服务员（服务端线程）  
**场 景**：客人去酒吧喝酒  
**BIO**：酒吧针对每个客人都安排一个服务员**全程服务他，直到他离开酒吧**。  
**NIO**：酒吧有一个或者几个服务员，还有一个总经理，**这个总经理会不停的主动的挨个像客人询问他是否需要服务**，如果需要就通知空闲的服务员来服务他，而如果没有空闲服务员则等待。  
**AIO**：酒吧也有一个或者几个服务员 **客人需要服务的时候喊他们服务**。

## 堵塞的本质 ##

了解了上面的通俗的例子后，我们需要思考，上诉的BIO问题在哪里，也就是堵塞式在哪一步堵塞的。  
客人去酒吧喝酒他也不会一直需要服务员服务他一样，服务员更多的还是在旁边等，这就是堵塞的本质，服务线程在等待。也就是：  
**客户端连接了服务端并不代表他就会立马请求数据，就算发送数据了也不是会一下就发送完成就立马处理的，而是会读满缓冲区或者在读到文件末尾（遇到：”/r”、”/n”、”/r/n”）才处理，而在网络较慢的时候可能客户端需要发送很久数据才会发送满缓冲区，在这过程中线程会一直堵塞的**。

## 专业的解释 ##

**BIO**：同步堵塞IO。服务端会针对每个客户端创建一个线程，这个线程在执行客户端请求处理的时候是堵塞的，不能响应其他请求。而服务端的总线程数是有限的，当线程不能再创建的时候，新的请求只有等原有的线程执行完毕才能再次执行这个请求。故这种模式势必会影响并发数以及造成资源的浪费。**这种模式更适合于少而长的连接**。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMzI5MzM0MzI_size_16_color_FFFFFF_t_70]  
**NIO**：同步非堵塞IO。服务端有一个轮巡者，该轮巡着会不停查看所有的客户端查看是否有请求数据且数据是否发送满了缓冲区，只有发送满了才会分配线程去执行这个请求，**但在JAVANIO中，这个轮询的动作是不需要代码类似于写一个while(true)来实现的，而只是调用了selector.select方法（可以看下面实现NIO的代码），其本质上是调用了操作系统的轮询机制，把轮询的动作交给了内核，查询到一个或者多个就返回。如果一个都没有还是会阻塞**。这一功能实际上是利用了系统内核中的一个叫**IO多路复用**的机制。然而，使用select函数的优点并不仅限于此。虽然上述方式允许单线程内处理多个IO请求，但是每个IO请求的过程还是阻塞的（在select函数上阻塞），平均时间甚至比同步阻塞IO模型还要长。如果用户线程只注册自己感兴趣的socket或者IO请求，然后去做自己的事情，等到数据到来时再进行处理，则可以提高CPU的利用率。NIO主要解决的问题是:  
处理大量并发连接中,只有少量活跃的,(就是很多人连接过来,但是他们都没有断开)只有少数是活跃的, 如果都活跃那NIO也没提高效率。**NIO方式适用于连接数目多且连接比较短（轻操作）的架构**，比如聊天服务器，并发局限于应用中，编程比较复杂，JDK1.4开始支持。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMzI5MzM0MzI_size_16_color_FFFFFF_t_70 1]  
—2019-2-18号更新  
一直以来，我都很疑问，为什么连接数很大的时候NIO依然可以保持很高的性能，它不需要去扫描所有的连接吗？性能不应该是线性下降的吗？直到我看到《Nett权威指南》中的这段话  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMzI5MzM0MzI_size_16_color_FFFFFF_t_70 2]

多路复用  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMzI5MzM0MzI_size_16_color_FFFFFF_t_70 3]  
多路复用采用的操作系统底层的模型，select poll epoll模型，现在大部分Linux系统采用epoll模式，相比于传统模式，这个能处理多个IO请求，传统模式会一直轮询是否数据准备好，浪费大量cpu，如果多个NIO，那么就多个线程一起轮询。所以引发出IO复用模式。而调用底层的select poll epoll等函数，让操作系统去轮询，查询到一个或者多个就返回。如果一个都没有还是会阻塞，本质上就是把轮询交给了内核。  
例如：Java NIO网络编程中，调用select方法，其本质上调用epoll\_wait返回准备好的IO。这个的好处在于一个单独线程就能调度很多IO请求业务。轮询交给了内核。

**AIO**：异步非堵塞IO。他的两个步骤处理是分开的，也就是说，一个连接可能他的数据接收是线程a完成的，数据处理是线程b完成的，他比BIO能处理更多请求，但是比不上NIO，但是他的处理性能又比BIO更差，因为一个连接他需要两次system call，而BIO只需要一次，所以这种IO模型应用的不多。“真正”的异步IO需要操作系统更强的支持。在IO多路复用模型中，事件循环将文件句柄的状态事件通知给用户线程，由用户线程自行读取数据、处理数据。而在异步IO模型中，当用户线程收到通知时，数据已经被内核读取完毕，并放在了用户线程指定的缓冲区内，内核在IO完成后通知用户线程直接使用即可。AIO方式使用于连接数目多且连接比较长（重操作）的架构，比如相册服务器，充分调用OS参与并发操作，编程比较复杂。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMzI5MzM0MzI_size_16_color_FFFFFF_t_70 4]  
如同所示，异步IO模型中，用户线程直接使用内核提供的异步IO API发起read请求，且发起后立即返回，继续执行用户线程代码。不过此时用户线程已经将调用的AsynchronousOperation和CompletionHandler注册到内核，然后操作系统开启独立的内核线程去处理IO操作。当read请求的数据到达时，由内核负责读取socket中的数据，并写入用户指定的缓冲区中。最后内核将read的数据和用户线程注册的CompletionHandler分发给内部Proactor，Proactor将IO完成的信息通知给用户线程（一般通过调用用户线程注册的完成事件处理函数），完成异步IO。

相比于IO多路复用模型，异步IO并不十分常用，不少高性能并发服务程序使用IO多路复用模型+多线程任务处理的架构基本可以满足需求。况且目前操作系统对异步IO的支持并非特别完善，更多的是采用IO多路复用模型模拟异步IO的方式（IO事件触发时不直接通知用户线程，而是将数据读写完毕后放到用户指定的缓冲区中）。AIO还有个缺点是接收数据需要预先分配缓存, 而不是NIO那种需要接收时才需要分配缓存, 所以对连接数量非常大但流量小的情况, 内存浪费很多。这也是为什么netty使用NIO而不使用AIO其中的原因。

## 代码 ##

篇幅有限，此处只列举关键代码，完整代码请移步文末下载，自己运行一遍，什么都懂了。

### BIO: ###

ServerNormal：

public final class ServerNormal { 
    	public synchronized static void start(int port) throws IOException{ 
            if(server != null) return;
            try{ 
                //通过构造函数创建ServerSocket
                //如果端口合法且空闲，服务端就监听成功
                server = new ServerSocket(port);
                System.out.println("服务器已启动，端口号：" + port);
                //通过无线循环监听客户端连接
                //如果没有客户端接入，将阻塞在accept操作上。
                while(true){ 
                    Socket socket = server.accept();
                    //当有新的客户端接入时，会执行下面的代码
                    //然后创建一个新的线程处理这条Socket链路
                    new Thread(new ServerHandler(socket)).start();
                }
            }finally{ 
                //一些必要的清理工作
                if(server != null){ 
                    System.out.println("服务器已关闭。");
                    server.close();
                    server = null;
                }
            }
        }
    }

ServerHandler:

public class ServerHandler implements Runnable{ 
    @Override
    public void run() { 
        BufferedReader in = null;
        PrintWriter out = null;
        in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
        out = new PrintWriter(socket.getOutputStream(),true);
        String expression;
        String result;
        while(true){ 
            //主要原因酒在于(expression = in.readLine())==null这句代码是堵塞的，时间都花在了等待上。
            if((expression = in.readLine())==null) break;
            System.out.println("服务器收到消息：" + expression);
        }
    }

可以看出来服务端是在客户端每次有新请求时都去启动了一个新线程的去接收客户端发送的数据的，BufferedReader.readLine()会读满缓冲区或者在读到文件末尾（遇到："/r"、"/n"、"/r/n"）才返回，这样就会导致网络很慢的适合线程可能会卡很久，从而导致服务端积聚大量的线程，大量的线程会严重影响服务器性能，甚至罢工。如何解决这个问题呢？NIO横空出世。

### NIO ###

ServerHandle

public class ServerHandle implements Runnable { 
        private Selector selector;
        private ServerSocketChannel serverChannel;
        private volatile boolean started;
    
        /** * 构造方法 * * @param port 指定要监听的端口号 */
        public ServerHandle(int port) { 
            try { 
                //创建选择器
                selector = Selector.open();
                //打开监听通道
                serverChannel = ServerSocketChannel.open();
                //如果为 true，则此通道将被置于阻塞模式；如果为 false，则此通道将被置于非阻塞模式
                //开启非阻塞模式
                serverChannel.configureBlocking(false);
                //绑定端口 backlog设为1024
                serverChannel.socket().bind(new InetSocketAddress(port), 1024);
                //监听客户端连接请求
                serverChannel.register(selector, SelectionKey.OP_ACCEPT);
                //标记服务器已开启
                started = true;
                System.out.println("服务器已启动，端口号：" + port);
            } catch (IOException e) { 
                e.printStackTrace();
                System.exit(1);
            }
        }
    
        public void stop() { 
            started = false;
        }
    
        @Override
        public void run() { 
            //循环遍历selector
            while (started) { 
                try { 
                    //无论是否有读写事件发生，selector每隔1s被唤醒一次
                    selector.select(10000);
                    //阻塞,只有当至少一个注册的事件发生的时候才会继续.
                    Set<SelectionKey> keys = selector.selectedKeys();
                    if (keys.size()>0) { 
                        System.out.println("哈哈哈，我接收到客户端的请求了"+keys.size());
                    }
                    for (Iterator<SelectionKey> i = keys.iterator(); i.hasNext();){ 
                        SelectionKey key = i.next();
                        i.remove();
                        try { 
                            handleInput(key);
                        } catch (Exception e) { 
                            if (key != null) { 
                                key.cancel();
                                if (key.channel() != null) { 
                                    key.channel().close();
                                }
                            }
                        }
                    }
                } catch (Throwable t) { 
                    t.printStackTrace();
                }
            }
            //selector关闭后会自动释放里面管理的资源
            if (selector != null) { 
                try { 
                    selector.close();
                } catch (Exception e) { 
                    e.printStackTrace();
                }
            }
        }
    
        private void handleInput(SelectionKey key) throws IOException { 
            if (key.isValid()) { 
                //处理新接入的请求消息
                if (key.isAcceptable()) { 
                    ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                    //通过ServerSocketChannel的accept创建SocketChannel实例
                    //完成该操作意味着完成TCP三次握手，TCP物理链路正式建立
                    SocketChannel sc = ssc.accept();
                    //设置为非阻塞的
                    sc.configureBlocking(false);
                    //注册为读
                    sc.register(selector, SelectionKey.OP_READ);
                }
                //读消息
                if (key.isReadable()) { 
                    SocketChannel sc = (SocketChannel) key.channel();
                    //创建ByteBuffer，并开辟一个1M的缓冲区
                    ByteBuffer buffer = ByteBuffer.allocate(1024);
                    //读取请求码流，返回读取到的字节数
                    int readBytes = sc.read(buffer);
                    //读取到字节，对字节进行编解码
                    if (readBytes > 0) { 
                        //将缓冲区当前的limit设置为position=0，用于后续对缓冲区的读取操作
                        buffer.flip();
                        //根据缓冲区可读字节数创建字节数组
                        byte[] bytes = new byte[buffer.remaining()];
                        //将缓冲区可读字节数组复制到新建的数组中
                        buffer.get(bytes);
                        String expression = new String(bytes, "UTF-8");
                        System.out.println("服务器收到消息：" + expression);
                        //处理数据
                        String result = null;
                        try { 
                            Thread.sleep(10000);
                            result = Calculator.cal(expression).toString();
                        } catch (Exception e) { 
                            result = "计算错误：" + e.getMessage();
                        }
                        //发送应答消息
                        doWrite(sc, result);
                    }
                    //没有读取到字节 忽略
    // else if(readBytes==0);
                    //链路已经关闭，释放资源
                    else if (readBytes < 0) { 
                        key.cancel();
                        sc.close();
                    }
                }
            }
        }
    
        //异步发送应答消息
        private void doWrite(SocketChannel channel, String response) throws IOException { 
            //将消息编码为字节数组
            byte[] bytes = response.getBytes();
            //根据数组容量创建ByteBuffer
            ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
            //将字节数组复制到缓冲区
            writeBuffer.put(bytes);
            //flip操作
            writeBuffer.flip();
            //发送缓冲区的字节数组
            channel.write(writeBuffer);
            //****此处不含处理“写半包”的代码
        }
    }

注意看selector.select方法，Selector是一个组件，可以检测多个NIO channel，看看读或者写事件是否就绪。多个Channel以事件的方式可以注册到同一个Selector，从而达到用一个线程处理多个请求成为可能。

列举NIO和BIO适合的场景。  
**BIO**：你有少量的连接使用非常高的带宽，一次发送大量的数据，也许典型的IO服务器实现可能非常契合。（因为这种方式服务端不需要启动太多的线程且设计简单，用NIO设计过于复杂）（就好像有一天酒吧来了一个酒神，喝酒是真滴快，如果采用NIO,那就总是需要让服务员跑来给他倒酒，人家也烦是不是？这种情况用BIO就要好很多，相当于专门给这位酒神客人分配一个服务员。）  
**NIO**：如果需要管理同时打开的成千上万个连接，这些连接每次只是发送少量的数据，例如聊天服务器，实现NIO的服务器可能是一个优势。（因为这种方式用BIO实现会造出服务端线程启动太多且因为数据较少不能得到有效的利用，而使用NIO则很适合这种场景）

用人可能会说，那你这个什么NIO不也还是堵塞的吗？你只是避免了服务端有过多的线程再等待啊，并没有实现异步啊。是的，本身上来说NIO并不是真正意义上的异步，称之为new io更合适。那要实现真正的异步怎么玩呢？那请看下面的AIO

### AIO ###

注意：aio是JDK1.7里面才新增的。aio即异步通知，也就是你写完数据了会异步调用一个方法，与NIO的区别在于NIO你需要等到服务端轮询到你，而AIO是你写完了你立马通知到客户端处理你的信息。  
AsyncServerHandler

public class AsyncServerHandler implements Runnable { 
        public CountDownLatch latch;
        public AsynchronousServerSocketChannel channel;
        @Override
        public void run() { 
            latch = new CountDownLatch(1);
            //用于接收客户端的连接，有客户端连接后会回调AcceptHandler里面的方法
            channel.accept(this,new AcceptHandler());
            //此处，让现场在此阻塞，防止服务端执行完成后退出
            latch.await();
        }
    }

AcceptHandler

public class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel, AsyncServerHandler> { 
        @Override
        public void completed(AsynchronousSocketChannel channel,AsyncServerHandler serverHandler) { 
            System.out.println("AcceptHandler.completed()");
            //继续接受其他客户端的请求
            Server.clientCount++;
            System.out.println("连接的客户端数：" + Server.clientCount);
            serverHandler.channel.accept(serverHandler, this);
            //创建新的Buffer
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            //异步读 第三个参数为接收消息回调的业务Handler
            channel.read(buffer, buffer, new ReadHandler(channel));
        }
        @Override
        public void failed(Throwable exc, AsyncServerHandler serverHandler) { 
            exc.printStackTrace();
            serverHandler.latch.countDown();
        }
    }

ReadHandler

public class ReadHandler implements CompletionHandler<Integer, ByteBuffer> { 
        //用于读取半包消息和发送应答
        private AsynchronousSocketChannel channel;
        public ReadHandler(AsynchronousSocketChannel channel) { 
            this.channel = channel;
        }
        //读取到消息后的处理
        @Override
        public void completed(Integer result, ByteBuffer attachment) { 
            System.out.println("ReadHandler.completed()");
            //flip操作
            attachment.flip();
            //根据
            byte[] message = new byte[attachment.remaining()];
            attachment.get(message);
            try { 
                String expression = new String(message, "UTF-8");
                System.out.println("服务器收到消息: " + expression);
                String calrResult = "发送给客户端的消息";
                //向客户端发送消息
                doWrite(calrResult);
            } catch (UnsupportedEncodingException e) { 
                e.printStackTrace();
            }
        }
        //发送消息
        private void doWrite(String result) { 
            byte[] bytes = result.getBytes();
            ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
            writeBuffer.put(bytes);
            writeBuffer.flip();
            //异步写数据 参数与前面的read一样
            channel.write(writeBuffer, writeBuffer,new CompletionHandler<Integer, ByteBuffer>() { 
                @Override
                public void completed(Integer result, ByteBuffer buffer) { 
                    //如果没有发送完，就继续发送直到完成
                    if (buffer.hasRemaining())
                        channel.write(buffer, buffer, this);
                    else{ 
                        //创建新的Buffer
                        ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                        //异步读 第三个参数为接收消息回调的业务Handler
                        channel.read(readBuffer, readBuffer, new ReadHandler(channel));
                    }
                }
                @Override
                public void failed(Throwable exc, ByteBuffer attachment) { 
                    try { 
                        channel.close();
                    } catch (IOException e) { 
                    }
                }
            });
        }
        @Override
        public void failed(Throwable exc, ByteBuffer attachment) { 
            try { 
                this.channel.close();
            } catch (IOException e) { 
                e.printStackTrace();
            }
        }
    }

回调的类需要实现CompletionHandler接口的两个方法。completed()：在成功适合调用，failed():在失败的适合调用.

可以看到所谓AIO关键的两个类是AsynchronousServerSocketChannel和AsynchronousSocketChannel。前者用于服务端，有个accept()方法，该方法用于接受客户端的连接，客户端有连接后会调用参数中的Handler类。注意这个方法是非堵塞的，只是在成功之后会调用。而AsynchronousSocketChannel的read(),write()均是非堵塞的，会在读取完了之后调用参数中的Handler类。基于这两个类就是真正意义上的异步了。

# 关于Netty #

可以看到java中的Nio编写是相对复杂的，而且还有诸如沾包和拆包的问题，因此，Netty应运而生，Netty简化了IO的编程方式，同时由于netty自定义协议等性能也会我们自己编写的程序要好，同时也能帮我们规避掉很多坑。

关于Netty的入门可以看笔我的另一篇[博文][Link 1]

关于本文的java源码下载：https://gitee.com/nuoqian1/IoTest

参考文章：  
https://ifeve.com/java-nio-vs-io/  
https://blog.csdn.net/anxpp/article/details/51512200  
https://blog.csdn.net/caolipeng\_918/article/details/49534113  
https://blog.csdn.net/happyzwh/article/details/53437570

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMzI5MzM0MzI_size_16_color_FFFFFF_t_70]: /images/20220525/a5276d6c45994e6f88979341a174e947.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMzI5MzM0MzI_size_16_color_FFFFFF_t_70 1]: /images/20220525/45b3145bf80d469fa92a5aa3f78114e1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMzI5MzM0MzI_size_16_color_FFFFFF_t_70 2]: /images/20220525/b63ebf8799ff4a8aae5f982652823ba0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMzI5MzM0MzI_size_16_color_FFFFFF_t_70 3]: /images/20220525/5a05dec00fa644c2ac3f04c6c32283ce.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMzI5MzM0MzI_size_16_color_FFFFFF_t_70 4]: /images/20220525/adf584fec9b440cda01030eb13430061.png
[Link 1]: https://blog.csdn.net/qq32933432/article/details/87721814