JAVA原生JDK网络编程 BIO/NIO

迈不过友情╰ 2022-12-27 04:44 154阅读 0赞

# 1 原生 JDK 网络编程 BIO #

传统的同步阻塞模型开发中，ServerSocket 负责绑定 IP 地址，启动监听端口；Socket 负责发起连接操作。连接成功后，双方通过输入和输出流进行同步阻塞式通信。传统 BIO 通信模型：采用 BIO 通信模型的服务端，通常由一个独立的 Acceptor 线程负责监听客户端的连接，它接收到客户端连接请求之后为每个客户端创建一个新的线程进行链路处理，处理完成后，通过输出流返回应答给客户端，线程销毁。即典型的一请求一应答模型，同时数据的读取写入也必须阻塞在一个线程内等待其完成。

该模型最大的问题就是缺乏弹性伸缩能力，当客户端并发访问量增加后，服务端的线程个数和客户端并发访问数呈 1:1 的正比关系，Java 中的线程也是比较宝贵的系统资源，线程数量快速膨胀后，系统的性能将急剧下降，随着访问量的继续增大，系统最终就 死- 掉- 了。为了改进这种一连接一线程的模型，我们可以使用线程池来管理这些线程，实现 1 个或多个线程处理 N 个客户端的模型（但是底层还是使用的同步阻塞 I/O），通常被称为“伪异步 I/O 模型“。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70]

public class BioServerPool { 
    
        private static ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);
    
        public static void main(String[] args) throws IOException { 
            ServerSocket serverSocket = new ServerSocket(8888);
            System.out.println("server is started......");
            while (true) { 
                executorService.submit(new ServerTask(serverSocket.accept()));
            }
        }
    
    
    
    
        /** * 服务端先实例化输入流，再实例话输出流，确保流通道的建立 */
        static class ServerTask implements Runnable { 
            private Socket socket;
    
            public ServerTask(Socket socket) { 
                this.socket = socket;
            }
    
            @Override
            public void run() { 
                System.out.println("ServerSocket accept socket");
                try (ObjectInputStream inputStream = new ObjectInputStream(socket.getInputStream());
                     ObjectOutputStream outputStream = new ObjectOutputStream(socket.getOutputStream())) { 
                    //接受消息
                    String message = inputStream.readUTF();
                    System.out.println(Thread.currentThread().getName() + " receive message: " + message);
                    //返回消息
                    outputStream.writeUTF("server success receive message");
                    outputStream.flush();
                } catch (Exception e) { 
                    e.printStackTrace();
                } finally { 
                    try { 
                        socket.close();
                    } catch (IOException e) { 
                        e.printStackTrace();
                    }
                }
            }
        }
    }

public class BioClient { 
    
        /** * 客户端先实例化输出流，再实例化输入流，确保流通道的建立 */
        public static void main(String[] args) throws IOException { 
    
    
            Socket socket = new Socket("127.0.0.1", 8888);
            try (ObjectOutputStream outputStream = new ObjectOutputStream(socket.getOutputStream());
                 ObjectInputStream inputStream = new ObjectInputStream(socket.getInputStream());) { 
                outputStream.writeUTF("Hello World");
                outputStream.flush();
                System.out.println("server response message :" + inputStream.readUTF());
            } catch (Exception e) { 
                e.printStackTrace();
            } finally { 
                socket.close();
            }
        }
    
    
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 2]

# 2 原生 JDK 网络编程- NIO #

## 2.1 什么是 NIO？ ##

NIO 库是在 JDK 1.4 中引入的。NIO 弥补了原来的 I/O 的不足，它在标准 Java 代码  
中提供了高速的、面向块的 I/O。NIO 翻译成 no-blocking io 或者 new io 都说得通。

## 2.2 和 BIO 的主要区别 ##

### (1) 面向流与面向缓冲 ###

Java NIO 和 IO 之间第一个最大的区别是，IO 是面向流的，NIO 是面向缓冲区的。Java IO面向流意味着每次从流中读一个或多个字节，直至读取所有字节，它们没有被缓存在任何地方。此外，它不能前后移动流中的数据。如果需要前后移动从流中读取的数据，需要先将它缓存到一个缓冲区。 Java NIO 的缓冲导向方法略有不同。数据读取到一个它稍后处理的缓冲区，需要时可在缓冲区中前后移动。这就增加了处理过程中的灵活性。但是，还需要检查是否该缓冲区中包含所有需要处理的数据。而且，需确保当更多的数据读入缓冲区时，不要覆盖缓冲区里尚未处理的数据。

### (2) 阻塞与非阻塞 IO ###

Java IO 的各种流是阻塞的。这意味着，当一个线程调用 read() 或 write()时，该线程被  
阻塞，直到有一些数据被读取，或数据完全写入。该线程在此期间不能再干任何事情了。  
Java NIO 的非阻塞模式，使一个线程从某通道发送请求读取数据，但是它仅能得到目  
前可用的数据，如果目前没有数据可用时，就什么都不会获取。而不是保持线程阻塞，所以直至数据变的可以读取之前，该线程可以继续做其他的事情。 非阻塞写也是如此。一个线程请求写入一些数据到某通道，但不需要等待它完全写入，这个线程同时可以去做别的事情。线程通常将非阻塞 IO 的空闲时间用于在其它通道上执行 IO 操作，所以一个单独的线程现在可以管理多个输入和输出通道（channel）。

## 2.3 NIO 三大核心组件 ##

NIO 有三大核心组件：Selector 选择器、Channel 管道、buffer 缓冲区。

(1) 一个线程对应一个Selector。  
(2) 一个Selector可以监听多个Channel。  
(3) 每个Channel对应一个Buffer。  
(4) 程序切换到哪个channel是由事件决定的。  
(5) Selector会根据不同的事件，在各个通道上切换。  
(6) Buffer是一个内存块（块IO比流IO效率高）。  
(7) BIO输入流、输出流是分开的，不能双向。Channel是双向的，Buffer也是双向的可读可写但是要flip切换。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 3]

## 2.4 Code ##

### (1) NioClient ###

public class NioClient { 
    
        private static NioClientHandle nioClientHandle;
        public static void main(String[] args) throws IOException { 
            start();
            Scanner scanner = new Scanner(System.in);
            while (sendMsg(scanner.next())) ;
        }
        //开启客户端
        private static void start() { 
            nioClientHandle = new NioClientHandle("127.0.0.1", 8000);
            new Thread(nioClientHandle, "client").start();
        }
        //发送消息
        public static boolean sendMsg(String msg) throws IOException { 
            nioClientHandle.sendMsg(msg);
            return true;
        }
    }

### (2) NioClientHandle ###

package com.io.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.SocketChannel;
    import java.util.Iterator;
    import java.util.Set;
    
    public class NioClientHandle implements Runnable { 
        private String host;
        private int port;
        private volatile boolean started;
        private Selector selector;
        private SocketChannel socketChannel;
        public NioClientHandle(String ip, int port) { 
            //1、绑定IP、端口
            this.host = ip;
            this.port = port;
            try { 
                //2、打开选择器和管道 并且定义为非阻塞模式
                selector = Selector.open();
                socketChannel = SocketChannel.open();
                socketChannel.configureBlocking(false);
                started = true;
            } catch (IOException e) { 
                e.printStackTrace();
            }
        }
        @Override
        public void run() { 
            try { 
                doConnect();
            } catch (IOException e) { 
                e.printStackTrace();
                System.exit(1);
            }
            while (started) { 
                try { 
                    //阻塞方法，没有事件返回的时候回阻塞
                    selector.select();
                    //拿到处理的事件和迭代器
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> it = selectionKeys.iterator();
                    //循环处理该事件
                    SelectionKey key = null;
                    while (it.hasNext()) { 
                        key = it.next();
                        handleInput(key);
                        it.remove();
                    }
                } catch (IOException e) { 
                    e.printStackTrace();
                }
            }
        }
        private void handleInput(SelectionKey key) throws IOException { 
            if (key.isValid()) { 
                //查看key和哪个管道绑定的
                SocketChannel sc = (SocketChannel) key.channel();
                //如果是连接事件，如果连接不成功那么直接退出
                if (key.isConnectable()) { 
                    if (!sc.finishConnect()) { 
                        System.exit(1);
                    }
                }
                if (key.isReadable()) { 
                    //定义 buffer
                    ByteBuffer buffer = ByteBuffer.allocate(1024);
                    //把channel的数据写入buffer
                    int read = sc.read(buffer);
                    if (read > 0) { 
                        buffer.flip();
                        byte[] bytes = new byte[buffer.remaining()];
                        buffer.get(bytes);
                        String rs = new String(bytes);
                        System.out.println("accept messasge: " + rs);
                    } else if (read < 0) { 
                        sc.close();
                        key.cancel();
                    }
                }
            }
        }
        private void doConnect() throws IOException { 
            //如果连接成功
            //如果连接不成功，注册一个连接事件
            if (socketChannel.connect(new InetSocketAddress(host, port))) { 
            } else { 
                socketChannel.register(selector, SelectionKey.OP_CONNECT);
            }
        }
        public void sendMsg(String msg) throws IOException { 
            socketChannel.register(selector, SelectionKey.OP_READ);
            byte[] bytes=msg.getBytes();
            ByteBuffer buffer=ByteBuffer.allocate(bytes.length);
            buffer.put(bytes);
            buffer.flip();
            socketChannel.write(buffer);
        }
    }

### (3) NioServer ###

package com.io.nio;
    
    public class NioServer { 
    
        private static NioServerHandle nioServerHandle;
    
        public static void start() { 
            nioServerHandle = new NioServerHandle(8000);
            new Thread(nioServerHandle, "Server").start();
        }
    
        public static void main(String[] args) { 
    
            start();
        }
    }

### (4) NioServerHandle ###

package com.io.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 NioServerHandle implements Runnable { 
        private Selector selector;
        private ServerSocketChannel serverSocketChannel;
        private volatile boolean started;
        public NioServerHandle(int port) { 
            try { 
                selector = Selector.open();
                serverSocketChannel = ServerSocketChannel.open();
                serverSocketChannel.configureBlocking(false);
                serverSocketChannel.socket().bind(new InetSocketAddress(port));
                serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
                started = true;
                System.out.println("服务器已启动，端口号：" + port);
            } catch (IOException e) { 
                e.printStackTrace();
            }
        }
        @Override
        public void run() { 
            //循环遍历selector
            while (started) { 
                try { 
                    //阻塞,只有当至少一个注册的事件发生的时候才会继续.
                    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 (Throwable t) { 
                    t.printStackTrace();
                }
            }
        }
        private void handleInput(SelectionKey key) throws IOException { 
            if (key.isValid()) { 
                //处理新接入的请求消息
                if (key.isAcceptable()) { 
                    //获得关心当前事件的channel
                    ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                    //通过ServerSocketChannel的accept创建SocketChannel实例
                    //完成该操作意味着完成TCP三次握手，TCP物理链路正式建立
                    SocketChannel sc = ssc.accept();
                    System.out.println("======socket channel 建立连接");
                    //设置为非阻塞的
                    sc.configureBlocking(false);
                    //连接已经完成了，可以开始关心读事件了
                    sc.register(selector, SelectionKey.OP_READ);
                }
                //读消息
                if (key.isReadable()) { 
                    System.out.println("======socket channel 数据准备完成，" + "可以去读==读取=======");
                    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 message = new String(bytes, "UTF-8");
                        System.out.println("服务器收到消息：" + message);
                        //处理数据
                        String result = "Hello," + message;
                        //发送应答消息
                        doWrite(sc, result);
                    }
                    //链路已经关闭，释放资源
                    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);
        }
    }

#### (5) 测试 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 4]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 5]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70]: /images/20221120/8d9810dce34b4f63a343bb4fd1ff0beb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 1]: /images/20221120/05e36828de9c4641a6e7fd67c5a6d4ce.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 2]: /images/20221120/8b980947a3c346e08276b57127fe45c8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 3]: /images/20221120/60f9f44c25ee4a5bbc0b899cced77d93.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 4]: /images/20221120/55d777b46ce74d6d84cafff2d9e708ec.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 5]: /images/20221120/a8be5e69898f446c893f79cd030dbd09.png