Java网络IO模型分析与实现

迈不过友情╰ 2023-09-26 18:42 129阅读 0赞

### 前言 ###

本篇文章会对**Java**中的网络**IO**模型的概念进行解释，并给出具体的**Java**代码实现，主要涉及如下部分。

1.  **BIO**（同步阻塞**IO**模型）的概念和**Java**编程实现；
2.  **Non Blocking IO**（同步非阻塞**IO**模型）的概念和**Java**编程实现；
3.  **IO**多路复用的概念；
4.  **NIO**（**New IO**）的概念和**Java**编程实现。

在开始本篇文章内容之前，有一个简单的关于**Socket**的知识需要说明：在进行网络通信的时候，需要一对**Socket**，一个运行于客户端，一个运行于服务端，同时服务端还会有一个服务端**Socket**，用于监听客户端的连接。下图进行一个简单示意。

![format_png][]

那么整个通信流程如下所示。

1.  服务端运行后，会在服务端创建**listen-socket**，**listen-socket**会绑定服务端的**ip**和**port**，然后服务端进入监听状态；
2.  客户端请求服务端时，客户端创建**connect-socket**，**connect-socket**描述了其要连接的服务端的**listen-socket**，然后**connect-socket**向**listen-socket**发起连接请求；
3.  **connect-socket**与**listen-socket**成功连接后（**TCP**三次握手成功），服务端会为已连接的客户端创建一个代表该客户端的**client-socket**，用于后续和客户端进行通信；
4.  客户端与服务端通过**socket**进行网络**IO**操作，此时就实现了客户端和服务端中的不同进程的通信。

需要知道的就是，在客户端与服务端通信的过程中，出现了三种**socket**，分别是。

1.  **listen-socket**。是服务端用于监听客户端建立连接的**socket**；
2.  **connect-socket**。是客户端用于连接服务端的**socket**；
3.  **client-socket**。是服务端监听到客户端连接请求后，在服务端生成的与客户端连接的**socket**。

（注：上述中的**socket**，可以被称为套接字，也可以被称为文件描述符。）

### 正文 ###

#### 一. BIO ####

**BIO**，即同步阻塞**IO**模型。用户进程调用**read**时发起**IO**操作，此时用户进程由用户态转换到内核态，只有在内核态中将**IO**操作执行完后，才会从内核态切换回用户态，这期间用户进程会一直阻塞。

**BIO**示意图如下。

![format_png 1][]

简单的**BIO**的**Java**编程实现如下。

**服务端实现**

public class BioServer {
    
        public static void main(String[] args) throws IOException {
            // 创建listen-socket
            ServerSocket listenSocket = new ServerSocket(8080);
            // 进入监听状态，是一个阻塞状态
            // 有客户端连接时从监听状态返回
            // 并创建代表这个客户端的client-socket
            Socket clientSocket = listenSocket.accept();
            // 获取client-socket输入流
            BufferedReader bufferedReader = new BufferedReader(
                    new InputStreamReader(clientSocket.getInputStream()));
            // 读取客户端发送的数据
            // 如果数据没准备好，会进入阻塞状态
            String data = bufferedReader.readLine();
            System.out.println(data);
    
            // 获取client-socket输出流
            BufferedWriter bufferedWriter = new BufferedWriter(
                    new OutputStreamWriter(clientSocket.getOutputStream()));
            // 服务端向客户端发送数据
            bufferedWriter.write("来自服务端的返回数据\n");
            // 刷新流
            bufferedWriter.flush();
        }
    
    }
    复制代码

**客户端实现**

public class BioClient {
    
        public static final String SERVER_IP = "127.0.0.1";
        public static final int SERVER_PORT = 8080;
    
        public static void main(String[] args) throws IOException {
            // 客户端创建connect-socket
            Socket connectSocket = new Socket(SERVER_IP, SERVER_PORT);
            // 获取connect-socket输出流
            BufferedWriter bufferedWriter = new BufferedWriter(
                    new OutputStreamWriter(connectSocket.getOutputStream()));
            // 客户端向服务端发送数据
            bufferedWriter.write("来自客户端的请求数据\n");
            // 刷新流
            bufferedWriter.flush();
    
            // 获取connect-socket输入流
            BufferedReader bufferedReader = new BufferedReader(
                    new InputStreamReader(connectSocket.getInputStream()));
            // 读取服务端发送的数据
            String returnData = bufferedReader.readLine();
            System.out.println(returnData);
        }
    
    }
    复制代码

**BIO**的问题就在于服务端在**accept**时是阻塞的，并且在主线程中，一次只能**accept**一个**Socket**，**accept**到**Socket**后，读取客户端数据时又是阻塞的。

#### 二. Non Blocking IO ####

**Non Blocking IO**，即同步非阻塞**IO**。是用户进程调用**read**时，用户进程由用户态转换到内核态后，此时如果没有系统资源数据能够被读取到内核缓冲区中，返回**read**失败，并从内核态切换回用户态。也就是用户进程发起**IO**操作后会立即得到一个操作结果。

**Non Blocking IO**示意图如下所示。

![format_png 2][]

简单的**Non Blocking IO**的**Java**编程实现如下。

public class NonbioServer {
    
        public static final List<SocketChannel> clientSocketChannels = new ArrayList<>();
    
        public static void main(String[] args) throws Exception {
            // 客户端创建listen-socket管道
            // 管道支持非阻塞模式和同时读写
            ServerSocketChannel listenSocketChannel = ServerSocketChannel.open();
            // 设置为非阻塞模式
            listenSocketChannel.configureBlocking(false);
            // 绑定监听的端口号
            listenSocketChannel.socket().bind(new InetSocketAddress(8080));
            // 在子线程中遍历clientSocketChannels并读取客户端数据
            handleSocketChannels();
    
            while (true) {
                // 非阻塞方式监听客户端连接
                // 如果无客户端连接则返回空
                // 有客户端连接则创建代表这个客户端的client-socket管道
                SocketChannel clientSocketChannel = listenSocketChannel.accept();
                if (clientSocketChannel != null) {
                    // 设置为非阻塞模式
                    clientSocketChannel.configureBlocking(false);
                    // 添加到clientSocketChannels中
                    // 用于子线程遍历并读取客户端数据
                    clientSocketChannels.add(clientSocketChannel);
                } else {
                    LockSupport.parkNanos(1000 * 1000 * 1000);
                }
            }
        }
    
        public static void handleSocketChannels() {
            new Thread(() -> {
                while (true) {
                    // 遍历每一个client-socket管道
                    Iterator<SocketChannel> iterator = clientSocketChannels.iterator();
                    while (iterator.hasNext()) {
                        SocketChannel clientSocketChannel = iterator.next();
                        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                        int read = 0;
                        try {
                            // 将客户端发送的数据读取到ByteBuffer中
                            // 这一步的操作也是非阻塞的
                            read = clientSocketChannel.read(byteBuffer);
                        } catch (IOException e) {
                            // 移除发生异常的client-socket管道
                            iterator.remove();
                            e.printStackTrace();
                        }
                        if (read == 0) {
                            System.out.println("客户端数据未就绪");
                        } else {
                            System.out.println("客户端数据为：" + new String(byteBuffer.array()));
                        }
                    }
                    LockSupport.parkNanos(1000 * 1000 * 1000);
                }
            }).start();
        }
    
    }
    复制代码

上述是**Non Blocking IO**的一个简单服务端的实现，相较于**BIO**，服务端在**accept**时是非阻塞的，在读取客户端数据时也是非阻塞的，但是还是存在如下问题。

1.  一次只能**accept**一个**Socket**；
2.  需要在用户进程中遍历所有的**SocketChannel**并调用**read()** 方法获取客户端数据，此时如果客户端数据未准备就绪，那么这一次的**read()** 操作的开销就是浪费的。

#### 三. IO多路复用 ####

在上述的**BIO**和**Non Blocking IO**中，一次系统调用，只会获取一个**IO**的状态，而如果采取**IO**多路复用机制，则可以一次系统调用获取多个**IO**的状态。

也就是获取多个**IO**的状态可以复用一次系统调用。

最简单的**IO**多路复用方式是基于**select**模型实现，步骤如下。

1.  在用户进程中将需要监控的**IO**文件描述符（**Socket**）注册到**IO**多路复用器中；
2.  执行**select**操作，此时用户进程由用户态转换到内核态（一次系统调用），然后在内核态中会轮询注册到**IO**多路复用器中的**IO**是否准备就绪，并得到所有准备就绪的**IO**的文件描述符列表，最后返回这些文件描述符列表；
3.  用户进程在**select**操作返回前会一直阻塞，直至**select**操作返回，此时用户进程就获得了所有就绪的**IO**的文件描述符列表；
4.  用户进程获得了就绪的**IO**的文件描述符列表后，就可以对这些**IO**进行相应的操作了。

换言之，**IO**多路复用中，只需要一次系统调用，**IO**多路复用器就可以告诉用户进程，哪些**IO**已经准备就绪可以进行操作了，而如果不采用**IO**多路复用，则需要用户进程自己遍历每个**IO**并调用**accept()** 或者**read()** 方法去判断，且一次**accept()** 或者**read()** 方法调用只能判断一个**IO**。

#### 四. NIO ####

**NIO**，即**New IO**。关于**NIO**，有如下三大组件。

1.  **channel**（管道）。介于**buffer**（字节缓冲区）和**Socket**（套接字）之间，用于数据的读写操作；
2.  **buffer**（字节缓冲区）。是用户程序和**channel**（管道）之间进行读写数据的中间区域；
3.  **selector**（**IO**多路复用器）。服务端的**listen-socket**和**client-socket**，客户端的**connect-socket**，都可以注册在**selector**上，注册的时候还需要指定监听的事件，比如为**listen-socket**指定监听的事件为**ACCEPT**事件，该事件发生则表示客户端建立了连接，还比如为**client-socket**指定监听的事件为**READ**事件，该事件发生则表示客户端发送的数据已经可读。

**NIO**的代码实现如下所示。

**服务端实现**

public class NioServer {
    
        private static Selector selector;
    
        public static void main(String[] args) {
    
            try {
                // 开启并得到多路复用器
                selector = Selector.open();
                // 服务端创建listen-socket管道
                ServerSocketChannel listenSocketChannel = ServerSocketChannel.open();
                // 设置为非阻塞模式
                listenSocketChannel.configureBlocking(false);
                // 为管道绑定端口
                listenSocketChannel.socket().bind(new InetSocketAddress(8080));
                // 将listen-socket管道注册到多路复用器上，并指定监听ACCEPT事件
                listenSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
    
                while (true) {
                    // 获取发生的事件，这个操作是阻塞的
                    selector.select();
                    // 拿到有事件发生的SelectionKey集合
                    // SelectionKey表示管道与多路复用器的绑定关系
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    // 遍历每个发生的事件，然后判断事件类型
                    // 根据事件类型，进行不同的处理
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    while (iterator.hasNext()) {
                        SelectionKey selectionKey = iterator.next();
                        iterator.remove();
                        if (selectionKey.isAcceptable()) {
                            // 处理客户端连接事件
                            handlerAccept(selectionKey);
                        } else if (selectionKey.isReadable()) {
                            // 处理客户端数据可读事件
                            handlerRead(selectionKey);
                        }
                    }
                    LockSupport.parkNanos(1000 * 1000 * 1000);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        private static void handlerAccept(SelectionKey selectionKey) {
            // 从事件中获取到listen-socket管道
            ServerSocketChannel listenSocketChannel = (ServerSocketChannel) selectionKey.channel();
            try {
                // 为连接的客户端创建client-socket管道
                SocketChannel clientSocketChannel = listenSocketChannel.accept();
                // 设置为非阻塞模式
                clientSocketChannel.configureBlocking(false);
                // 将client-socket管道注册到多路复用器上，并指定监听READ事件
                clientSocketChannel.register(selector, SelectionKey.OP_READ);
                // 给客户端发送数据
                clientSocketChannel.write(ByteBuffer.wrap("连接已建立\n".getBytes()));
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        private static void handlerRead(SelectionKey selectionKey) {
            // 从事件中获取到client-socket管道
            SocketChannel clientSocketChannel = (SocketChannel) selectionKey.channel();
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            try {
                // 读取客户端数据
                int read = clientSocketChannel.read(byteBuffer);
                if (read <= 0) {
                    // 关闭管道
                    clientSocketChannel.close();
                    // 从多路复用器移除绑定关系
                    selectionKey.cancel();
                } else {
                    System.out.println(new String(byteBuffer.array()));
                }
            } catch (IOException e1) {
                try {
                    // 关闭管道
                    clientSocketChannel.close();
                } catch (IOException e2) {
                    e2.printStackTrace();
                }
                // 从多路复用器移除绑定关系
                selectionKey.cancel();
                e1.printStackTrace();
            }
        }
    
    }
    复制代码

**客户端实现**

public class NioClient {
    
        private static Selector selector;
    
        public static final String SERVER_IP = "127.0.0.1";
        public static final int SERVER_PORT = 8080;
    
        public static void main(String[] args) {
            try {
                // 开启并得到多路复用器
                selector = Selector.open();
                // 创建connect-socket管道
                SocketChannel connectSocketChannel = SocketChannel.open();
                // 设置为非阻塞模式
                connectSocketChannel.configureBlocking(false);
                // 设置服务端IP和端口
                connectSocketChannel.connect(new InetSocketAddress(SERVER_IP, SERVER_PORT));
                // 将connect-socket管道注册到多路复用器上，并指定监听CONNECT事件
                connectSocketChannel.register(selector, SelectionKey.OP_CONNECT);
    
                while (true) {
                    // 获取发生的事件，这个操作是阻塞的
                    selector.select();
                    // 拿到有事件发生的SelectionKey集合
                    // SelectionKey表示管道与多路复用器的绑定关系
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    // 遍历每个发生的事件，然后判断事件类型
                    // 根据事件类型，进行不同的处理
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    while (iterator.hasNext()) {
                        SelectionKey selectionKey = iterator.next();
                        iterator.remove();
                        if (selectionKey.isConnectable()) {
                            // 处理连接建立事件
                            handlerConnect(selectionKey);
                        } else if (selectionKey.isReadable()) {
                            // 处理服务端数据可读事件
                            handlerRead(selectionKey);
                        }
                    }
                    LockSupport.parkNanos(1000 * 1000 * 1000);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        private static void handlerConnect(SelectionKey selectionKey) throws IOException {
            // 拿到connect-socket管道
            SocketChannel connectSocketChannel = (SocketChannel) selectionKey.channel();
            if (connectSocketChannel.isConnectionPending()) {
                connectSocketChannel.finishConnect();
            }
            // 设置为非阻塞模式
            connectSocketChannel.configureBlocking(false);
            // 将connect-socket管道注册到多路复用器上，并指定监听READ事件
            connectSocketChannel.register(selector, SelectionKey.OP_READ);
            // 向服务端发送数据
            connectSocketChannel.write(ByteBuffer.wrap("客户端发送的数据\n".getBytes()));
        }
    
        private static void handlerRead(SelectionKey selectionKey) {
            // 拿到connect-socket管道
            SocketChannel connectSocketChannel = (SocketChannel) selectionKey.channel();
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            try {
                // 读取服务端数据
                int read = connectSocketChannel.read(byteBuffer);
                if (read <= 0) {
                    // 关闭管道
                    connectSocketChannel.close();
                    // 从多路复用器移除绑定关系
                    selectionKey.cancel();
                } else {
                    System.out.println(new String(byteBuffer.array()));
                }
            } catch (IOException e1) {
                try {
                    // 关闭管道
                    connectSocketChannel.close();
                } catch (IOException e2) {
                    e2.printStackTrace();
                }
                // 从多路复用器移除绑定关系
                selectionKey.cancel();
                e1.printStackTrace();
            }
        }
    
    }
    复制代码

### 总结 ###

本篇文章中所讨论的**IO**模型，都是同步**IO**模型，而何谓同步异步，何谓阻塞非阻塞，可以总结如下。

1.  **同步**和**异步**。**同步IO**表示需要在用户进程中主动的去询问操作系统数据是否准备好，如果没有准备好，还需要持续的询问，直到数据准备好为止；而**异步IO**则是在用户进程中只需要询问操作系统一次，后续数据准备好后操作系统会主动的将数据给到用户进程；
2.  **阻塞**和**非阻塞**。**阻塞IO**就是发起一次系统调用后，会一直阻塞直到有结果返回；而**非阻塞IO**就是发起一次系统调用后，会立即得到一个返回结果。

实际上在**Java**中是有对异步**IO**（**AIO**）做支持，但是**AIO**依赖操作系统的底层实现，而目前**Linux**对**AIO**的支持不成熟，所以**AIO**的使用并不多，像主流的网络应用框架**Netty**也都没有使用到**AIO**。

[format_png]: https://img-blog.csdnimg.cn/img_convert/3ae62bec09d3360919055589985fd929.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/c25ec9cda483c925cba98e23ade1f7e6.webp?x-oss-process=image/format,png
[format_png 2]: https://img-blog.csdnimg.cn/img_convert/094a1b3839945661f9e364eb0ab75052.webp?x-oss-process=image/format,png