Java 中 3 种常见的 IO 模型

浅浅的花香味﹌ 2022-10-08 00:49 135阅读 0赞

## 知识背景 ##

操作系统：

*  为了保证操作系统的稳定性和安全性，一个进程的地址空间被分为 `用户空间` 和 `内核空间` ；
 *  用户空间不能直接访问内核空间，要想访问必须进行 `系统调用` ；
 *  `IO` 操作只有内核空间才能完成，所以用户进程需要进行系统调用；
 *  所以用户空间仅仅是发起系统调用请求，真正的 `IO` 操作执行是由内核空间完成的。

常见的 `IO` 模型：

*  同步阻塞 IO ⭐
 *  同步非阻塞 IO
 *  IO 多路复用 ⭐
 *  信号驱动 IO
 *  异步 IO ⭐

其中带有星号的模型为 java 中常见的 3 种模型，下面将分别介绍。

## BIO ##

BIO 即 `Blocking I/O`；字面意思就可以看出它属于同步阻塞 IO。

如下图，应用程序发出一个 read 调用，内核空间需要经历准备数据的几个阶段，准备好之后返回数据给应用程序。期间如果另一个应用程序也需要 read 调用，那么它必须等待；这就是阻塞。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzk0MTM2NA_size_16_color_FFFFFF_t_70_pic_center][]

BIO 最大的特点就是一次只能处理一个调用，这在高并发的场景下肯定是不行的。

### 示例代码 ###

客户端：

public class IOClient { 
    
      public static void main(String[] args) { 
        // 创建多个线程，模拟多个客户端连接服务端
        new Thread(() -> { 
          try { 
            Socket socket = new Socket("127.0.0.1", 3333);
            while (true) { 
              try { 
                socket.getOutputStream().write((new Date() + ": hello world").getBytes());
                Thread.sleep(2000);
              } catch (Exception e) { 
              }
            }
          } catch (IOException e) { 
          }
        }).start();
      }
    }

服务端：

public class IOServer { 
    
      public static void main(String[] args) throws IOException { 
        // TODO 服务端处理客户端连接请求
        ServerSocket serverSocket = new ServerSocket(3333);
    
        // 接收到客户端连接请求之后为每个客户端创建一个新的线程进行链路处理
        new Thread(() -> { 
          while (true) { 
            try { 
              // 阻塞方法获取新的连接
              Socket socket = serverSocket.accept();
              
              // 每一个新的连接都创建一个线程，负责读取数据
              new Thread(() -> { 
                try { 
                  int len;
                  byte[] data = new byte[1024];
                  InputStream inputStream = socket.getInputStream();
                  // 按字节流方式读取数据
                  while ((len = inputStream.read(data)) != -1) { 
                    System.out.println(new String(data, 0, len));
                  }
                } catch (IOException e) { 
                }
              }).start();
            } catch (IOException e) { 
            }
          }
        }).start();
      }
    }

## NIO ##

NIO 就是 `Non-blocking I/O`。字面翻译为非阻塞，但其实它是属于多路复用模型。下面就非阻塞模型和多路复用模型作简要区分：

*  非阻塞模型：关键词是 `轮询` ，例如小明需要找人帮忙，于是找到张三，第一次张三在忙，第二次张三还在忙，此后小明的做法是每一个小时来一次，直到等到张三有空为止。该做法很不明智，具体体现在浪费了小明的时间，来来回回都是需要消耗处理器资源的。
 *  多路复用模型：还是小明需要帮忙，不过这次多了一个查询系统，这个系统可以提供谁有空，小明经过查询发现 3 个好朋友当中只有李四有空，于是找了李四帮忙。这就避免了浪费处理器资源。

![多路复用][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzk0MTM2NA_size_16_color_FFFFFF_t_70_pic_center 1]  
如图，在多路复用模型中，线程想获得内核空间的数据，必须先发起 `select` 系统调用来询问内核空间是否有空；当内核空间有空时会回复应用程序一个 `ready` 。应用程序得知内核空间准备就绪之后就会再次发送 `read` 调用来请求数据。

*  `select` 系统调用：内核提供的系统调用，它支持一次查询多个系统调用的可用状态。几乎所有的操作系统都支持。
 *  这里的 select 调用相当于上例中的查询系统；ready 相当于查到了李四有空。

IO 多路复用模型最大的特点是通过减少无效的系统调用，减少了对 CPU 资源的消耗。

> Java 中的 NIO 于 Java 1.4 中引入，对应 java.nio 包，提供了 `Channel` , `Selector`，`Buffer` 等抽象。其中有一个非常重要的选择器 ( `Selector` ) 的概念，也可以被称为 **多路复用器**。通过它，只需要一个线程（单线程）便可以管理多个客户端连接。只有当客户端数据到了之后，才会为其服务。

![多路复用器原理图][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzk0MTM2NA_size_16_color_FFFFFF_t_70_pic_center 2]

这里 `Selector` 选择器的作用是监听多个通道的状态，判断是否空闲。

使用单线程管理的理由是，从操作系统的角度来看，切换线程开销是比较昂贵的，并且每个线程都需要占用系统资源，因此暂用线程越少越好。

### 示例代码 ###

客户端：

服务端：

public class NIOServer { 
      public static void main(String[] args) throws IOException { 
        // 1. serverSelector负责轮询是否有新的连接，服务端监测到新的连接之后，不再创建一个新的线程，
        // 而是直接将新连接绑定到clientSelector上，这样就不用 IO 模型中 1w 个 while 循环在死等
        Selector serverSelector = Selector.open();
        // 2. clientSelector负责轮询连接是否有数据可读
        Selector clientSelector = Selector.open();
    
        new Thread(() -> { 
          try { 
            // 对应IO编程中服务端启动
            ServerSocketChannel listenerChannel = ServerSocketChannel.open();
            listenerChannel.socket().bind(new InetSocketAddress(3333));
            listenerChannel.configureBlocking(false);
            listenerChannel.register(serverSelector, SelectionKey.OP_ACCEPT);
    
            while (true) { 
              // 监测是否有新的连接，这里的1指的是阻塞的时间为 1ms
              if (serverSelector.select(1) > 0) { 
                Set<SelectionKey> set = serverSelector.selectedKeys();
                Iterator<SelectionKey> keyIterator = set.iterator();
    
                while (keyIterator.hasNext()) { 
                  SelectionKey key = keyIterator.next();
    
                  if (key.isAcceptable()) { 
                    try { 
                      // (1) 每来一个新连接，不需要创建一个线程，而是直接注册到clientSelector
                      SocketChannel clientChannel = ((ServerSocketChannel) key.channel()).accept();
                      clientChannel.configureBlocking(false);
                      clientChannel.register(clientSelector, SelectionKey.OP_READ);
                    } finally { 
                      keyIterator.remove();
                    }
                  }
                }
              }
            }
          } catch (IOException ignored) { 
          }
        }).start();
        new Thread(() -> { 
          try { 
            while (true) { 
              // (2) 批量轮询是否有哪些连接有数据可读，这里的1指的是阻塞的时间为 1ms
              if (clientSelector.select(1) > 0) { 
                Set<SelectionKey> set = clientSelector.selectedKeys();
                Iterator<SelectionKey> keyIterator = set.iterator();
    
                while (keyIterator.hasNext()) { 
                  SelectionKey key = keyIterator.next();
    
                  if (key.isReadable()) { 
                    try { 
                      SocketChannel clientChannel = (SocketChannel) key.channel();
                      ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                      // (3) 面向 Buffer
                      clientChannel.read(byteBuffer);
                      byteBuffer.flip();
                      System.out.println(
                          Charset.defaultCharset().newDecoder().decode(byteBuffer).toString());
                    } finally { 
                      keyIterator.remove();
                      key.interestOps(SelectionKey.OP_READ);
                    }
                  }
                }
              }
            }
          } catch (IOException ignored) { 
          }
        }).start();
      }
    }

## AIO ##

AIO 即 `Asynchronous I/O`，也可以称之为 NIO 2。Java 7 中引入，它是异步 IO 模型。

![异步模型][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzk0MTM2NA_size_16_color_FFFFFF_t_70_pic_center 3]  
异步 IO 是基于事件和回调机制实现的，也就是说应用请求之后会直接返回，不会阻塞在那里，当后台处理完成，操作系统会通知响应的线程进行后续的操作。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzk0MTM2NA_size_16_color_FFFFFF_t_70_pic_center]: /images/20221005/0ee160bb477b4a93815f98437f136875.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzk0MTM2NA_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221005/766ae80b431844ebac393c67cbef0379.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzk0MTM2NA_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221005/5c8fe1a4834a4879821841969fabde72.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzk0MTM2NA_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221005/b76d1782b4f64d80a4401fb5a84b2b92.png