Java bio nio aio

我就是我 2024-04-17 06:00 68阅读 0赞

面试被问了相关的东西，回答的很模糊。于是打算了解这方面知识，demo一下。  
从这个博客学到了一些东西

> https://blog.csdn.net/anxpp/article/details/51512200\#commentBox

bio，又叫同步阻塞IO，服务器实现模式为，当一个请求到来服务端，服务端就启动一个新线程进行处理，如果这个连接不做任何事情就会造成不必要的开销，当客户端连接数过多时，造成服务器死掉。

nio,又叫同步非阻塞IO。服务器实现模式为，当客户端一个请求到服务器端时，客户端的所有请求都会注册到多路复用器上，多路复用器轮询到连接有IO请求时，才会启动一个线程进行处理。

aio,异步非阻塞IO。异步非阻塞IO，服务器实现模式为客户端一个有效的请求，都由OS处理完服务器再去启动线程去处理。

bio模式:  
![在这里插入图片描述][format_png]  
demo一下：bio 大概的思路

public final class ServerNormal {    // 服务端代码
        //默认的端口号
        private static int DEFAULT_PORT = 8083;
        //单例的ServerSocket
        private static ServerSocket server;
        //根据传入参数设置监听端口，如果没有参数调用以下方法并使用默认值
        public static void start() throws IOException {
            //使用默认值
            start(DEFAULT_PORT);
        }
        //这个方法不会被大量并发访问，不太需要考虑效率，直接进行方法同步就行了
    
        public  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链路
                    System.out.println("准备创建线程");
                    new Thread(new ServerHandler(socket)).start();
                }
            }finally{
                //一些必要的清理工作
                if(server != null){
                    System.out.println("服务器已关闭。");
                    server.close();
                    server = null;
                }
            }
        }
    
        public static void main(String[] args) {
    
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try{
    
                        ServerNormal.start();
    
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }

package com.hnist.lzn.BIO;
    
    import java.io.BufferedReader;
    import java.io.IOException;
    import java.io.InputStreamReader;
    import java.io.PrintWriter;
    import java.net.Socket;
    
    public class ServerHandler implements Runnable {
    
        private Socket socket;
        public ServerHandler(Socket socket) {
            this.socket = socket;
        }
        @Override
        public void run() {
            BufferedReader in = null;
            PrintWriter out = null;
            try {
                in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
                out = new PrintWriter(socket.getOutputStream(), true);
                String expression;
                String result;
                while (true) {
                    //通过BufferedReader读取一行
                    //如果已经读到输入流尾部，返回null,退出循环
                    //如果得到非空值，就尝试计算结果并返回
                    System.out.println("等待消息");
                    try {
                        // in.readLine()一直在阻塞,当有流到来，expression有值才会跳
                    while((expression = in.readLine())!=null)
                    {
                        System.out.println("服务器收到消息：" + expression);
                        out.println("服务器收到");
                    }
                    } catch (Exception e) {
                       e.printStackTrace();
                    }
    
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                //一些必要的清理工作
                if (in != null) {
                    try {
                        in.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    in = null;
                }
                if (out != null) {
                    out.close();
                    out = null;
                }
                if (socket != null) {
                    try {
                        socket.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    socket = null;
                }
            }
        }
    }

package com.hnist.lzn.BIO;
    
    import java.io.BufferedReader;
    import java.io.IOException;
    import java.io.InputStreamReader;
    import java.io.PrintWriter;
    import java.net.Socket;
    import java.util.Scanner;
    
    public class Client {
    
        //默认的端口号
        private static int DEFAULT_SERVER_PORT = 8083;
        private static String DEFAULT_SERVER_IP = "127.0.0.1";
       // public static void send(String expression){
           // send(DEFAULT_SERVER_PORT,expression);
       // }
        public static void send(){
           // System.out.println("算术表达式为：" + expression);
    
            Socket socket = null;
            BufferedReader in = null;
            PrintWriter out = null;
            Scanner scanner = new Scanner(System.in);
            try{
                socket = new Socket(DEFAULT_SERVER_IP,DEFAULT_SERVER_PORT);
                in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
    
                out = new PrintWriter(socket.getOutputStream(),true);
    
                while(true) {
    
                    System.out.println("请输入：");
                    out.println(scanner.nextLine());
                    System.out.println("___结果为：" + in.readLine());
                }
            }catch(Exception e){
                e.printStackTrace();
            }finally{
                //一下必要的清理工作
                if(in != null){
                    try {
                        in.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    in = null;
                }
                if(out != null){
                    out.close();
                    out = null;
                }
                if(socket != null){
                    try {
                        socket.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    socket = null;
                }
            }
        }
    
        public static void main(String[] args) {
    
            new Thread(new Runnable() {
                @Override
                public void run() {
                        Client.send();
                }
            }).start();
    
    
        }
    
    
    }

效果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTEzNzY5OQ_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTEzNzY5OQ_size_16_color_FFFFFF_t_70 1]

nio:同步非阻塞IO，关键采用了事件驱动思想来实现一个多路转换器。  
nio与bio最大的区别是只需要开启一个线程就可以处理来自多个客户端的事件，这是怎么做的呢？  
就是多路复用器，可以监听来自多个客户端的IO事件：  
如果服务端监听到客户端连接请求，便为其建立通信套接字，然后继续返回监听，若同时有多个客户端连接请求到来也可以全部做到，依次为它们建立连接。  
若服务端监听到来自已经创建了套接字的客户端发来的数据，就会调用相应的接口来处理，若同时有多个客户端发来数据也可以依次进行处理。  
监听多个客户端的连接请求和接受数据请求同时还能监听自己有时候有数据发送。

适用场景：  
NIO适合处理连接数目较多，但是连接比较短，操作，zookeeper,jetty都是nio  
BIO方式适用于连接数目比较小且固定的场景，这种方式对服务器资源要求比较高，并发局限于应用中。

nio提供了与传统BIO模型中的Socket和ServerSocket相对应的SocketChannel和ServerSocketChannel两种不同的套接字管道实现。

新增的2种通道都支持阻塞和非阻塞两种模式。  
阻塞模式就和适用传统的支持一样，比较简单，性能和可靠性不好，非阻塞正好相反。  
对于低负载，低并发程序，使用同步阻塞I/O来提升开发速率和更好的维护性，对于高负载，高并发的网络应用，应用NIO的非阻塞模式来开发。

首先介绍一下：  
1.Buffer缓冲区  
buffer是一个对象，所有数据都是用缓冲区处理的。在读取数据时，直接读取到缓冲区；在写入数据时，也是写入到缓冲区中，任何时候访问NIO中的数据，都是通过缓冲区进行操作。  
缓冲区实际是一个数组，并提供了对数据结构化访问以及维护读写位置等信息。  
具体的缓冲区有：ByteBuffer,CharBuffer,ShortBuffer,IntBuffer,LongBuffer,FloatBuffer,DoubbleBuffer.都实现了相同的接口：Buffer.  
2.通道Channel  
我们对数据的读写和写入要通过Channel,它就像水管一样，是一个通道。通道不同于流的地方就是通道是双向的。可以用于读，写，和同时读写操作。  
底层的操作系统的通道一般都是全双工的，所以全双工的Channel比流能更好的映射底层的操作系统API。  
Channel主要分2大类：  
SelectableChannel:用户网络读写  
FileChannel:文件操作  
后面涉及的ServerSocketChannel和SocketChannel都是SelectableChannel的子类。

3.Selector 多路复用器  
Selector是Java NIO编程的基础。  
Selector提供选择已经就绪的任务的能力：Selector会不断轮询注册在其上的Channel,如果某个Channe上面发生读或写事件，这个Channel处于就绪状态，会被Selector轮询出来，然后通过SelectionKey可以获取Channel的集合，进行后续的I/O操作。

一个Selector可以同时轮询多个Channel,因为JDK使用epoll()代替传统的select实现，所以没有最大连接数限制。所以只需要一个线程负责Selector的轮询，就可以接入成千上万的客户端。  
![在这里插入图片描述][format_png 1]

package com.hnist.lzn.NIO;
    
    public class Server {
        private static int DEFAULT_PORT = 12345;
        private static ServerHandle serverHandle;
        public static void start(){
            start(DEFAULT_PORT);
        }
        public static  void start(int port){
            if(serverHandle!=null)
                serverHandle.stop();
            serverHandle = new ServerHandle(port);
            new Thread(serverHandle,"Server").start();
        }
        public static void main(String[] args){
            start();
        }
    
    }

package com.hnist.lzn.NIO;
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.ServerSocketChannel;
    import java.nio.channels.SocketChannel;
    import java.util.Iterator;
    import java.util.Set;
    
    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(1000);
                    //阻塞,只有当至少一个注册的事件发生的时候才会继续.
    //				selector.select();
                    // select key  监听了io流,只要有流过来，才会进行下面处理
                    Set<SelectionKey> keys = selector.selectedKeys();
                    Iterator<SelectionKey> it = keys.iterator();
                    SelectionKey key = null;
                    while(it.hasNext()){
                        key = it.next();
                        it.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()){
                System.out.println("handleInput 方法处理，监听信息过来");
                //处理新接入的请求消息, 被连接才能执行,新请求，才进去，ssc,被注册过了，不会再注册。但是会监听
                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);
    
                    System.out.println("确实被阻塞....");
                }
                //读消息 ,只有能读，才能进去
                if(key.isReadable()){
                    System.out.println("客户端还未发送消息，就能读吗？...应该被阻塞。。。");
                    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 = "服务器传回消息";
    
                        //发送应答消息
                        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);
            //****此处不含处理“写半包”的代码
        }
    
    }

package com.hnist.lzn.NIO;
    
    public class Client {
        private static String DEFAULT_HOST = "127.0.0.1";
        private static int DEFAULT_PORT = 12345;
        private static ClientHandle clientHandle;
        public static void start(){
            start(DEFAULT_HOST,DEFAULT_PORT);
        }
        public static  void start(String ip,int port){
            if(clientHandle!=null)
                clientHandle.stop();
            clientHandle = new ClientHandle(ip,port);
            new Thread(clientHandle,"Server").start();
        }
        //向服务器发送消息
        public static boolean sendMsg(String msg) throws Exception{
            if(msg.equals("q")) return false;
            clientHandle.sendMsg(msg);
            return true;
        }
        public static void main(String[] args){
            start();
        }
    
    }

package com.hnist.lzn.NIO;
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.net.SocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.SocketChannel;
    import java.util.Iterator;
    import java.util.Scanner;
    import java.util.Set;
    
    public class ClientHandle implements Runnable {
    
        private String host;
        private int port;
        private Selector selector;
        private SocketChannel socketChannel;
        private volatile boolean started;
        private SocketAddress address;
        Scanner scanner = new Scanner(System.in);
    
        public ClientHandle(String ip,int port) {
            this.host = ip;
            this.port = port;
            try{
                address = new InetSocketAddress(host,port);
                //创建选择器
                selector = Selector.open();
                //打开监听通道
                socketChannel = SocketChannel.open(address);
                //如果为 true，则此通道将被置于阻塞模式；如果为 false，则此通道将被置于非阻塞模式
                socketChannel.configureBlocking(false);//开启非阻塞模式
                started = true;
            }catch(IOException e){
                e.printStackTrace();
                System.exit(1);
            }
        }
        public void stop(){
            started = false;
        }
        @Override
        public void run() {
            try{
                doConnect();
                System.out.println("客户端连接..");
            }catch(IOException e){
                e.printStackTrace();
                System.exit(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            String str = scanner.nextLine();
            //循环遍历selector
            while(started){
                try{
                    sendMsg(str);
                    //无论是否有读写事件发生，selector每隔1s被唤醒一次
                    selector.select(1000);
                    //阻塞,只有当至少一个注册的事件发生的时候才会继续.
    //				selector.select();
                    Set<SelectionKey> keys = selector.selectedKeys();
                    Iterator<SelectionKey> it = keys.iterator();
                    SelectionKey key = null;
                    while(it.hasNext()){
                        key = it.next();
                        it.remove();
                        try{
                            handleInput(key);
                        }catch(Exception e){
                            if(key != null){
                                key.cancel();
                                if(key.channel() != null){
                                    key.channel().close();
                                }
                            }
                        }
                    }
                }catch(Exception e){
                    e.printStackTrace();
                    System.exit(1);
                }
                str = scanner.nextLine();
            }
            //selector关闭后会自动释放里面管理的资源
            if(selector != null)
                try{
                    selector.close();
                }catch (Exception e) {
                    e.printStackTrace();
                }
        }
        private void handleInput(SelectionKey key) throws Exception {
            if(key.isValid()){
                System.out.println("有效...");
                SocketChannel sc = (SocketChannel) key.channel();
                if(key.isConnectable()){
                    System.out.println("连接...");
                    if(sc.finishConnect());
                    else System.exit(1);
                }
                //
                //sendMsg(scanner.nextLine());
                //doWrite(sc,scanner.nextLine());
               // sc.register(this.selector,SelectionKey.OP_READ);
                //读消息
                if(key.isReadable()){
                    System.out.println("能读..");
                    //创建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 result = new String(bytes,"UTF-8");
                        System.out.println("客户端收到消息：" + result);
                    }
                    //没有读取到字节 忽略
    //				else if(readBytes==0);
                    //链路已经关闭，释放资源
                    else if(readBytes<0){
                        key.cancel();
                        sc.close();
                    }
                }
            }
        }
        //异步发送消息
        private void doWrite(SocketChannel channel,String request) throws IOException{
            //将消息编码为字节数组
            byte[] bytes = request.getBytes();
            //根据数组容量创建ByteBuffer
            ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
            //将字节数组复制到缓冲区
            writeBuffer.put(bytes);
            //flip操作
            writeBuffer.flip();
            //发送缓冲区的字节数组
            channel.write(writeBuffer);
    
            //****此处不含处理“写半包”的代码
        }
        private void doConnect() throws IOException, InterruptedException {
    
            //else
            socketChannel.register(selector, SelectionKey.OP_READ);
    
        }
        public void sendMsg(String msg) throws Exception{
          //  socketChannel.register(selector, SelectionKey.OP_READ);
            doWrite(socketChannel, msg);
        }
    
    }

这里客户端之前的代码有点问题，后来按着下列博客改了下

> https://blog.csdn.net/canot/article/details/51372651

实现：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTEzNzY5OQ_size_16_color_FFFFFF_t_70 2]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTEzNzY5OQ_size_16_color_FFFFFF_t_70 3]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTEzNzY5OQ_size_16_color_FFFFFF_t_70 4]

[format_png]: https://imgconvert.csdnimg.cn/aHR0cDovL2Jsb2cuYW54cHAuY29tL3Vzci91cGxvYWRzLzIwMTYvMDUvNTQ5NTIwOTE2LnBuZw?x-oss-process=image/format,png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTEzNzY5OQ_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/d1a8efb2212449c39da80d096cf8802d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTEzNzY5OQ_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/9a14fdc257ec42488dfd807f16620122.png
[format_png 1]: https://imgconvert.csdnimg.cn/aHR0cDovL3N0YXRpYy5vc2NoaW5hLm5ldC91cGxvYWRzL2ltZy8yMDE1MTAvMjMwOTQ1MjhfT0Y5Yy5qcGc?x-oss-process=image/format,png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTEzNzY5OQ_size_16_color_FFFFFF_t_70 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/588f3bbf22b14d97a1e212daf175e621.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTEzNzY5OQ_size_16_color_FFFFFF_t_70 3]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/26552a94c3c448aab3ac16718ea5a5d3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTEzNzY5OQ_size_16_color_FFFFFF_t_70 4]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/b8ca6ea743f740acb096101a8f1d5d9a.png