网络编程之IO模型

2023-12-05 14:07 由 DaFanJoy 发表于 #后端开发

我们讨论网络编程中的IO模型时，需要先明确什么是IO以及IO操作为什么在程序开发中是很关键的一部分，首先我们看下IO的定义。

IO的定义

IO操作（Input/Output操作）是计算机系统中的一种重要操作，用于数据的输入和输出，通常涉及到计算机与外部设备（如硬盘、网卡、键盘、鼠标、打印机等）之间的数据传输和交互的都可以认为是IO操作。

IO操作可以分为两种主要类型：

1 输入操作（Input）指从外部设备或数据源中读取数据到计算机内存或程序当中。例如从硬盘读取文件内容、从键盘接收用户的输入、从网卡接收数据等.

2 输出操作（Output）指将计算机内存中的数据写入到外部设备或数据目标中。例如将数据写入到硬盘上的文件、文字打印输出、将数据发送到网络上等都属于输出操作。

无论是哪种I/O 操作分为两个部分：

数据准备，将数据加载到内核缓存。（数据加载到操作系统）
将内核缓存中的数据加载到用户缓存（从操作系统复制到应用中）

![[Pasted image 20231014111531.png]]

因此开发工作当中涉及到的网络读写、数据库操作、文件操作、日志打印的，都可以归为IO操作，IO操作的性能可以受到多种因素的影响，包括硬件性能、操作系统的优化、文件系统的性能等等。而网络IO特指计算机程序与网络之间进行数据交互的过程，在编码层面我们可以简单把它定义成的Socket套接字操作，如Socket的创建和关闭，数据的发送和接收。

IO的重要性

那么为什么我们要关心IO操并对应存在多种IO模型呢，因为IO操作的本质是要与硬件进行数据交互，这个过程是需要时间的，返回的结果也是需要等待的，从而也就造成了我们经常说的阻塞问题，你的程序是需要停在那里等待的，所以在大部分程序开发场景中，IO操作通常是最耗时的操作之一，并且最容易成为性能瓶颈的关键因素，特别是一旦你的程序开始上规模的，同样的一段程序，一旦需要处理的数量级上去，就会升级成一个复杂的问题；同理网络IO之所以重要且被人反复提及，是因为网络开发特别是服务端的开发中不可能处理的是单一链接，处理100个链接与处理100万个链接面对的性能挑战是不能同日而语的，这也是C10K问题的由来，所以一些需要处理海量链接的服务应用，比如IOT物联网服务，推送服务，为了提高IO操作的性能，通常会使用一些阻塞、非阻塞IO、异步IO模型来优化IO对整个应用的性能影响，也就是我们经常听到的BIO、NIO、AIO等IO模型，当然这只是解决上面所说问题的方案其中的一个环节。

IO依据阻塞，非阻塞，同步，异步等特点可以划分为阻塞IO(BIO)、非阻塞IO(NIO)、多路复用IO(multiplexing IO)、异步IO(AIO)。每一种IO都有他们的使用场景和优势，其中平常我们说的NIO已经包含了IO的多路复用，以下这张图各个IO和阻塞非阻塞，同步异步之间的关系。

其中阻塞、非阻塞、多路IO复用这些需要轮询处理的都是同步IO，真正的异步IO中程序只需要等待一个完成的信号的通知，也就是我们通常说的异步回调机制。所以拉一个子线程去轮训或使用select、poll、epoll都不是异步。

网络编程IO模型

上面我们阐述了IO的定义以及重要性，现在结合Java代码的具体实现看下不同IO模型的具体实现与特点。

1、阻塞IO模型

BIO（Blocking I/O）：
- BIO是最传统的阻塞I/O模型，意味着当一个线程执行I/O操作时，它会一直等待直到操作完成。
- 每个I/O操作都需要一个独立的线程来处理，这会导致线程数量的大幅增加，降低了系统的并发性能。
- 适用于一般连接数不多的应用场景。

public class BioServer {  
      
    public static void main(String[] args) throws IOException, InterruptedException {  
         
       ExecutorService executor = new ThreadPoolExecutor(2, 4, 1000, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10),  
             new ThreadPoolExecutor.CallerRunsPolicy());  
       ServerSocket serverSocket = new ServerSocket();  
       serverSocket.bind(new InetSocketAddress("127.0.0.1",9091));//绑定IP地址与端口，定义一个服务端socket,开启监听  
       while (true) {  
          Socket socket = serverSocket.accept();//这里如果没有客户端链接，会一直阻塞等待  
          //Thread.sleep(1000*30);  
          executor.execute(new BioServerHandler(socket));  
            
       }  
    }  
}

public class BioServerHandler implements Runnable{  
  
    private final Socket socket;  
      
    public BioServerHandler(Socket socket) {  
       this.socket=socket;  
    }  
      
    @Override  
    public void run() {  
       // TODO Auto-generated method stub  
       try {  
            
          while (true) {  
             byte[] rbytes = new byte[1024];  
             InputStream inputStream =  socket.getInputStream(); //通过IO输入流接受消息  
                int rlength=inputStream.read(rbytes, 0, 1024); //消息长度  
             byte[] bytes = new byte[rlength];  
             System.arraycopy(rbytes, 0, bytes, 0, rlength);  
             String message = new String(bytes);  
             System.out.printf("Client: %s%n", message);  
  
             PrintStream writer = new PrintStream(socket.getOutputStream()); //通过IO输出流发送消息  
             writer.println("Hello BIO Client");  
          }  
  
            
       } catch (IOException e) {  
          // TODO Auto-generated catch block  
          e.printStackTrace();  
       }  
         
    }  
  
}

2、 NIO模型

NIO 和多路复用是密切相关的概念，它们通常结合使用，特别是在网络编程中NIO 通常通过多路复用机制来实现非阻塞 I/O。多路复用允许一个线程同时监视多个通道的状态，当某个通道有数据可读或可写时，多路复用机制会通知应用程序，这使得应用程序能够高效地处理多个通道的 I/O 事件。

NIO引入了Channel和Buffer的概念，允许一个线程管理多个Channel，从而提高了系统的并发性能。

Channel和Buffer：NIO引入了Channel和Buffer的概念，允许一个线程管理多个Channel。Channel表示与数据源（如文件或网络套接字）的连接，而Buffer是用于读取或写入数据的缓冲区。线程可以将数据从Channel读取到Buffer，或者将数据从Buffer写入到Channel，而无需等待数据准备好。
Selector（选择器）：Selector是NIO的核心组件之一，它允许一个线程同时管理多个Channel。Selector可以检测多个Channel是否准备好读取或写入数据，从而使线程能够非阻塞地等待数据准备好的通知。这种机制称为事件驱动，允许一个线程同时监视多个通道上的事件，只处理已经准备好的事件，而不是等待每个通道的数据准备好。
非阻塞调用：NIO中的Channel和Socket通常可以配置为非阻塞模式。在非阻塞模式下，当进行读取或写入操作时，如果没有数据准备好，不会阻塞线程，而是立即返回一个状态码，表示没有数据可用，这样线程可以继续处理其他Channel，而不会被一个阻塞的操作阻塞。

综合上述机制，NIO允许一个线程同时处理多个连接，并在数据准备好时进行处理，而不会阻塞等待数据准备好。这种基于底层事件驱动的方式提高了应用系统的并发性能，特别适用于需要处理大量连接的场景。

public class NioServer {  
    public static void main(String[] args) {  
       try {  
          // TODO Auto-generated method stub  
            // 1.获取Selector选择器  
            Selector selector = Selector.open();  
            // 2.获取通道  
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();  
            // 3.设置为非阻塞  
            serverSocketChannel.configureBlocking(false);  
            // 4.绑定连接  
            serverSocketChannel.bind(new InetSocketAddress("127.0.0.1", 8091));  
            // 5.将通道注册到选择器上,并注册的操作为：“接收”操作  
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);  
              
              
            // 6.采用轮询的方式，查询获取“准备就绪”的注册过的操作  
            while (true) {  
                selector.select(); //阻塞
                Set<SelectionKey> selectionKeys = selector.selectedKeys();  
                Iterator<SelectionKey> iterator = selectionKeys.iterator();  
                while (iterator.hasNext()) {  
                    SelectionKey selectionKey = iterator.next();  
                    new NioServerHandler(serverSocketChannel, selectionKey).handle();  
                }  
               selectionKeys.clear();  
            }  
              
              
       } catch (Exception e) {  
          // TODO: handle exception  
       }  
         
  
    }  
  
}

public class NioServerHandler {  
      
    private ServerSocketChannel serverSocketChannel;  
      
    private SelectionKey selectionKey;  
      
    public  NioServerHandler(ServerSocketChannel serverSocketChannel, SelectionKey selectionKey) {  
       this.serverSocketChannel=serverSocketChannel;  
       this.selectionKey=selectionKey;  
    }  
  
    public void handle() {  
        ByteBuffer inputBuff = ByteBuffer.allocate(1024); // 分配读ByteBuffer  
        ByteBuffer outputBuff = ByteBuffer.allocate(1024); // 分配写ByteBuffer  
        try {  
            if (selectionKey.isAcceptable()) { //链接事件  
                SocketChannel socketChannel = serverSocketChannel.accept();  
                if (socketChannel == null) {  
                    return;  
                }  
                socketChannel.configureBlocking(false);  
                socketChannel.register(selectionKey.selector(), SelectionKey.OP_READ);  
            }  
  
            if (selectionKey.isReadable()) {//读事件  
                SocketChannel socketChannel = (SocketChannel) selectionKey.channel();  
                if (socketChannel == null) {  
                    return;  
                }  
  
                inputBuff.clear();  
                int length = socketChannel.read(inputBuff);  
                if (length == -1) {  
                    socketChannel.close();  
                    selectionKey.cancel();  
                    return;  
                }  
  
                inputBuff.flip();  
               
             byte[] bytes = new byte[length];  
             System.arraycopy(inputBuff.array(), 0, bytes, 0, length);  
                  
                System.err.println( BytesUtils.toHexString(bytes));  
  
  
                socketChannel.register(selectionKey.selector(), SelectionKey.OP_WRITE);  
                selectionKey.selector().wakeup();//唤醒选择器  
            }  
  
            if (selectionKey.isWritable()) {//写事件  
                outputBuff.clear();  
                SocketChannel socketChannel = (SocketChannel) selectionKey.channel();  
                if (socketChannel == null) {  
                    return;  
                }  
                String message = "Hello, Client. " + UUID.randomUUID();  
                System.err.println(message);  
                outputBuff.put(message.getBytes(StandardCharsets.UTF_8));  
                outputBuff.flip();  
                socketChannel.write(outputBuff);  
  
                socketChannel.register(selectionKey.selector(), SelectionKey.OP_READ);  
                selectionKey.selector().wakeup();//唤醒选择器  
            }  
        } catch (Exception e) {  
            e.printStackTrace();  
        }  
    }  
  
}

2、 AIO模型

AIO（Asynchronous I/O），也称为异步 I/O，是一种用于处理输入和输出操作的编程模型。它与传统的BIO和 NIO 模型有一些显著的不同之处。它不需要像BIO与NIO通过轮询方式去检查数据是否准备好，而是由操作系统完成。当数据准备好后，操作系统会通知应用程序，并在回调函数中进行处理。

AIO使用了三个核心组件：AsynchronousChannel、CompletionHandler和
AsynchronousServerSocketChannel。其中，AsynchronousChannel是读/写数据的通道，CompletionHandler是I/O操作完成时的回调方法，AsynchronousServerSocketChannel是异步服务器端套接字通道，用于监听客户端的连接请求。
以下是 AIO 的主要特点：

异步操作：AIO 的最重要特点是异步操作。在 AIO 模型中，应用程序发起 I/O 操作后，不需要等待操作完成，而是可以继续执行其他任务。当操作完成时，操作系统会通知应用程序，这种模型不会阻塞应用程序的执行。
回调机制：AIO 使用回调机制来处理 I/O 完成事件。应用程序在发起异步操作时需要提供一个回调函数或回调对象，用于处理操作完成后的事件通知。这使得 AIO 编程更具事件驱动的特性。
提高并发性能：AIO 能够在高并发环境中提供更好的性能，因为它在NIO允许一个线程管理多个 I/O 操作的基础上实现了异步回调，因此可以更好地处理大量连接。
复杂性：AIO 编程相对复杂，因为它涉及到回调和状态管理。编写和维护 AIO 代码可能需要更多的工作，但可以提供更好的性能和响应性。

总之AIO 是一种适合高并发、低延迟要求的应用程序的编程模型。它的异步特性和事件驱动的方式使得应用程序能够更好地利用系统资源，提供更好的性能和响应性。然而AIO通常比传统BIO或 NIO的具体实现更复杂，因此设计和编码的复杂度会相对较高。

public class AioServer {  
  
    static ExecutorService executor = new ThreadPoolExecutor(2, 4, 1000, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10),  
            new ThreadPoolExecutor.CallerRunsPolicy());  
    public static void main(String[] args) throws IOException, InterruptedException {  
        AsynchronousServerSocketChannel serverChannel = AsynchronousServerSocketChannel.open();  
        serverChannel.bind(new InetSocketAddress("127.0.0.1", 8091));  
  
            serverChannel.accept(null, new CompletionHandler<AsynchronousSocketChannel, Void>() {  
                @Override  
                public void completed(AsynchronousSocketChannel clientChannel, Void attachment) {  
                    // 继续接受下一个连接  
                    serverChannel.accept(null, this);  
  
                    executor.execute(new AioServerHandler(clientChannel));  
                }  
  
                @Override  
                public void failed(Throwable exc, Void attachment) {  
                    exc.printStackTrace();  
                }  
            });  
  
            Thread.currentThread().join();  
        }  
          
}

public class AioServerHandler implements Runnable {  
  
    private AsynchronousSocketChannel clientChannel;  
    public  AioServerHandler(AsynchronousSocketChannel clientChannel){  
        this.clientChannel=clientChannel;  
    }  
    @Override  
    public void run() {  
        ByteBuffer inputBuff = ByteBuffer.allocate(1024); // 分配读ByteBuffer  
        ByteBuffer outputBuff = ByteBuffer.allocate(1024); // 分配写ByteBuffer  
        clientChannel.read(inputBuff, null, new CompletionHandler<Integer, Void>() {  
            @Override  
            public void completed(Integer bytesRead, Void attachment) {  
                if (bytesRead == -1) {  
                    // 客户端关闭连接  
                    try {  
                        clientChannel.close();  
                    } catch (Exception e) {  
                        e.printStackTrace();  
                    }  
                    return;  
                }  
  
                inputBuff.flip();  
                byte[] data = new byte[bytesRead];  
                inputBuff.get(data);  
                String message = new String(data);  
                System.out.println("Received message: " + message);  
  
  
                outputBuff.clear();  
  
                outputBuff.put(message.getBytes(StandardCharsets.UTF_8));  
                outputBuff.flip();  
                clientChannel.write(outputBuff, null, new CompletionHandler<Integer, Void>() {  
                    @Override  
                    public void completed(Integer result, Void attachment) {  
                        System.out.println("Sent response: " + message);  
                    }  
  
                    @Override  
                    public void failed(Throwable exc, Void attachment) {  
                        exc.printStackTrace();  
                    }  
                });  
  
                inputBuff.clear();  
                clientChannel.read(inputBuff, null, this);  
            }  
  
            @Override  
            public void failed(Throwable exce, Void attachment) {  
                exce.printStackTrace();  
                try {  
                    clientChannel.close();  
                } catch (Exception e) {  
                    e.printStackTrace();  
                }  
            }  
        });  
    }  
}

结语

了解不同IO模型的特点与具体实现是我们设计与开发大型应用程序的基础，有助于我们根据不同的应用场景与性能要求选用最适合的模型并进行架构设计；同时一些常用的网络服务框架如Mina与Netty也都基于高性能的IO模型进行了深度的封装与扩展实现，我们可以通过结合这些框架的源码加深对网络IO模型在实际开发中应用的理解。后续我也会基于本章内容继续丰富完善，阐述一个完整的网络应用程序需要具备的各种功能并进行实现。

github地址：https://github.com/dafanjoy/jcode