ToB企服应用市场:ToB评测及商务社交产业平台

标题: 网络编程之IO模型 [打印本页]
作者: 海哥    时间: 2024-2-3 10:43
标题: 网络编程之IO模型
我们讨论网络编程中的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模型

  1. public class BioServer {  
  2.       
  3.     public static void main(String[] args) throws IOException, InterruptedException {  
  4.          
  5.        ExecutorService executor = new ThreadPoolExecutor(2, 4, 1000, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10),  
  6.              new ThreadPoolExecutor.CallerRunsPolicy());  
  7.        ServerSocket serverSocket = new ServerSocket();  
  8.        serverSocket.bind(new InetSocketAddress("127.0.0.1",9091));//绑定IP地址与端口，定义一个服务端socket,开启监听  
  9.        while (true) {  
  10.           Socket socket = serverSocket.accept();//这里如果没有客户端链接，会一直阻塞等待  
  11.           //Thread.sleep(1000*30);  
  12.           executor.execute(new BioServerHandler(socket));  
  13.             
  14.        }  
  15.     }  
  16. }
复制代码
  1. public class BioServerHandler implements Runnable{  
  2.   
  3.     private final Socket socket;  
  4.       
  5.     public BioServerHandler(Socket socket) {  
  6.        this.socket=socket;  
  7.     }  
  8.       
  9.     @Override  
  10.     public void run() {  
  11.        // TODO Auto-generated method stub  
  12.        try {  
  13.             
  14.           while (true) {  
  15.              byte[] rbytes = new byte[1024];  
  16.              InputStream inputStream =  socket.getInputStream(); //通过IO输入流接受消息  
  17.                 int rlength=inputStream.read(rbytes, 0, 1024); //消息长度  
  18.              byte[] bytes = new byte[rlength];  
  19.              System.arraycopy(rbytes, 0, bytes, 0, rlength);  
  20.              String message = new String(bytes);  
  21.              System.out.printf("Client: %s%n", message);  
  22.   
  23.              PrintStream writer = new PrintStream(socket.getOutputStream()); //通过IO输出流发送消息  
  24.              writer.println("Hello BIO Client");  
  25.           }  
  26.   
  27.             
  28.        } catch (IOException e) {  
  29.           // TODO Auto-generated catch block  
  30.           e.printStackTrace();  
  31.        }  
  32.          
  33.     }  
  34.   
  35. }
复制代码
2、 NIO模型

NIO 和多路复用是密切相关的概念，它们通常结合使用，特别是在网络编程中NIO 通常通过多路复用机制来实现非阻塞 I/O。多路复用允许一个线程同时监视多个通道的状态，当某个通道有数据可读或可写时，多路复用机制会通知应用程序，这使得应用程序能够高效地处理多个通道的 I/O 事件。
NIO引入了Channel和Buffer的概念，允许一个线程管理多个Channel，从而提高了系统的并发性能。
综合上述机制，NIO允许一个线程同时处理多个连接，并在数据准备好时进行处理，而不会阻塞等待数据准备好。这种基于底层事件驱动的方式提高了应用系统的并发性能，特别适用于需要处理大量连接的场景。
  1. public class NioServer {  
  2.     public static void main(String[] args) {  
  3.        try {  
  4.           // TODO Auto-generated method stub  
  5.             // 1.获取Selector选择器  
  6.             Selector selector = Selector.open();  
  7.             // 2.获取通道  
  8.             ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();  
  9.             // 3.设置为非阻塞  
  10.             serverSocketChannel.configureBlocking(false);  
  11.             // 4.绑定连接  
  12.             serverSocketChannel.bind(new InetSocketAddress("127.0.0.1", 8091));  
  13.             // 5.将通道注册到选择器上,并注册的操作为：“接收”操作  
  14.             serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);  
  15.               
  16.               
  17.             // 6.采用轮询的方式，查询获取“准备就绪”的注册过的操作  
  18.             while (true) {  
  19.                 selector.select(); //阻塞
  20.                 Set<SelectionKey> selectionKeys = selector.selectedKeys();  
  21.                 Iterator<SelectionKey> iterator = selectionKeys.iterator();  
  22.                 while (iterator.hasNext()) {  
  23.                     SelectionKey selectionKey = iterator.next();  
  24.                     new NioServerHandler(serverSocketChannel, selectionKey).handle();  
  25.                 }  
  26.                selectionKeys.clear();  
  27.             }  
  28.               
  29.               
  30.        } catch (Exception e) {  
  31.           // TODO: handle exception  
  32.        }  
  33.          
  34.   
  35.     }  
  36.   
  37. }
复制代码
  1. public class NioServerHandler {  
  2.       
  3.     private ServerSocketChannel serverSocketChannel;  
  4.       
  5.     private SelectionKey selectionKey;  
  6.       
  7.     public  NioServerHandler(ServerSocketChannel serverSocketChannel, SelectionKey selectionKey) {  
  8.        this.serverSocketChannel=serverSocketChannel;  
  9.        this.selectionKey=selectionKey;  
  10.     }  
  11.   
  12.     public void handle() {  
  13.         ByteBuffer inputBuff = ByteBuffer.allocate(1024); // 分配读ByteBuffer  
  14.         ByteBuffer outputBuff = ByteBuffer.allocate(1024); // 分配写ByteBuffer  
  15.         try {  
  16.             if (selectionKey.isAcceptable()) { //链接事件  
  17.                 SocketChannel socketChannel = serverSocketChannel.accept();  
  18.                 if (socketChannel == null) {  
  19.                     return;  
  20.                 }  
  21.                 socketChannel.configureBlocking(false);  
  22.                 socketChannel.register(selectionKey.selector(), SelectionKey.OP_READ);  
  23.             }  
  24.   
  25.             if (selectionKey.isReadable()) {//读事件  
  26.                 SocketChannel socketChannel = (SocketChannel) selectionKey.channel();  
  27.                 if (socketChannel == null) {  
  28.                     return;  
  29.                 }  
  30.   
  31.                 inputBuff.clear();  
  32.                 int length = socketChannel.read(inputBuff);  
  33.                 if (length == -1) {  
  34.                     socketChannel.close();  
  35.                     selectionKey.cancel();  
  36.                     return;  
  37.                 }  
  38.   
  39.                 inputBuff.flip();  
  40.                
  41.              byte[] bytes = new byte[length];  
  42.              System.arraycopy(inputBuff.array(), 0, bytes, 0, length);  
  43.                   
  44.                 System.err.println( BytesUtils.toHexString(bytes));  
  45.   
  46.   
  47.                 socketChannel.register(selectionKey.selector(), SelectionKey.OP_WRITE);  
  48.                 selectionKey.selector().wakeup();//唤醒选择器  
  49.             }  
  50.   
  51.             if (selectionKey.isWritable()) {//写事件  
  52.                 outputBuff.clear();  
  53.                 SocketChannel socketChannel = (SocketChannel) selectionKey.channel();  
  54.                 if (socketChannel == null) {  
  55.                     return;  
  56.                 }  
  57.                 String message = "Hello, Client. " + UUID.randomUUID();  
  58.                 System.err.println(message);  
  59.                 outputBuff.put(message.getBytes(StandardCharsets.UTF_8));  
  60.                 outputBuff.flip();  
  61.                 socketChannel.write(outputBuff);  
  62.   
  63.                 socketChannel.register(selectionKey.selector(), SelectionKey.OP_READ);  
  64.                 selectionKey.selector().wakeup();//唤醒选择器  
  65.             }  
  66.         } catch (Exception e) {  
  67.             e.printStackTrace();  
  68.         }  
  69.     }  
  70.   
  71. }
复制代码
2、 AIO模型

AIO（Asynchronous I/O），也称为异步 I/O，是一种用于处理输入和输出操作的编程模型。它与传统的BIO和 NIO 模型有一些显著的不同之处。它不需要像BIO与NIO通过轮询方式去检查数据是否准备好，而是由操作系统完成。当数据准备好后，操作系统会通知应用程序，并在回调函数中进行处理。
AIO使用了三个核心组件：AsynchronousChannel、CompletionHandler和
AsynchronousServerSocketChannel。其中，AsynchronousChannel是读/写数据的通道，CompletionHandler是I/O操作完成时的回调方法，AsynchronousServerSocketChannel是异步服务器端套接字通道，用于监听客户端的连接请求。
以下是 AIO 的主要特点：
总之AIO 是一种适合高并发、低延迟要求的应用程序的编程模型。它的异步特性和事件驱动的方式使得应用程序能够更好地利用系统资源，提供更好的性能和响应性。然而AIO通常比传统BIO或 NIO的具体实现更复杂，因此设计和编码的复杂度会相对较高。
  1. public class AioServer {  
  2.   
  3.     static ExecutorService executor = new ThreadPoolExecutor(2, 4, 1000, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10),  
  4.             new ThreadPoolExecutor.CallerRunsPolicy());  
  5.     public static void main(String[] args) throws IOException, InterruptedException {  
  6.         AsynchronousServerSocketChannel serverChannel = AsynchronousServerSocketChannel.open();  
  7.         serverChannel.bind(new InetSocketAddress("127.0.0.1", 8091));  
  8.   
  9.             serverChannel.accept(null, new CompletionHandler<AsynchronousSocketChannel, Void>() {  
  10.                 @Override  
  11.                 public void completed(AsynchronousSocketChannel clientChannel, Void attachment) {  
  12.                     // 继续接受下一个连接  
  13.                     serverChannel.accept(null, this);  
  14.   
  15.                     executor.execute(new AioServerHandler(clientChannel));  
  16.                 }  
  17.   
  18.                 @Override  
  19.                 public void failed(Throwable exc, Void attachment) {  
  20.                     exc.printStackTrace();  
  21.                 }  
  22.             });  
  23.   
  24.             Thread.currentThread().join();  
  25.         }  
  26.          
  27. }
复制代码
  1. public class AioServerHandler implements Runnable {  
  2.   
  3.     private AsynchronousSocketChannel clientChannel;  
  4.     public  AioServerHandler(AsynchronousSocketChannel clientChannel){  
  5.         this.clientChannel=clientChannel;  
  6.     }  
  7.     @Override  
  8.     public void run() {  
  9.         ByteBuffer inputBuff = ByteBuffer.allocate(1024); // 分配读ByteBuffer  
  10.         ByteBuffer outputBuff = ByteBuffer.allocate(1024); // 分配写ByteBuffer  
  11.         clientChannel.read(inputBuff, null, new CompletionHandler<Integer, Void>() {  
  12.             @Override  
  13.             public void completed(Integer bytesRead, Void attachment) {  
  14.                 if (bytesRead == -1) {  
  15.                     // 客户端关闭连接  
  16.                     try {  
  17.                         clientChannel.close();  
  18.                     } catch (Exception e) {  
  19.                         e.printStackTrace();  
  20.                     }  
  21.                     return;  
  22.                 }  
  23.   
  24.                 inputBuff.flip();  
  25.                 byte[] data = new byte[bytesRead];  
  26.                 inputBuff.get(data);  
  27.                 String message = new String(data);  
  28.                 System.out.println("Received message: " + message);  
  29.   
  30.   
  31.                 outputBuff.clear();  
  32.   
  33.                 outputBuff.put(message.getBytes(StandardCharsets.UTF_8));  
  34.                 outputBuff.flip();  
  35.                 clientChannel.write(outputBuff, null, new CompletionHandler<Integer, Void>() {  
  36.                     @Override  
  37.                     public void completed(Integer result, Void attachment) {  
  38.                         System.out.println("Sent response: " + message);  
  39.                     }  
  40.   
  41.                     @Override  
  42.                     public void failed(Throwable exc, Void attachment) {  
  43.                         exc.printStackTrace();  
  44.                     }  
  45.                 });  
  46.   
  47.                 inputBuff.clear();  
  48.                 clientChannel.read(inputBuff, null, this);  
  49.             }  
  50.   
  51.             @Override  
  52.             public void failed(Throwable exce, Void attachment) {  
  53.                 exce.printStackTrace();  
  54.                 try {  
  55.                     clientChannel.close();  
  56.                 } catch (Exception e) {  
  57.                     e.printStackTrace();  
  58.                 }  
  59.             }  
  60.         });  
  61.     }  
  62. }
复制代码
结语

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






github地址：https://github.com/dafanjoy/jcode关注微信公众号，查看更多技术文章。



免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！



欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4