聊聊分布式架构——BIO到NIO的演进

I/O

I/O模型

BIO示例

BIO与NIO比较

NIO的三大核心

NIO核心之缓冲区

Buffer常用子类：

Buffer常用API

Buffer中的重要概念

NIO核心之通道

FileChannel 类

FileChannel常用方法

NIO核心之选择器

概述

应用

NIO非阻塞原理分析

服务端流程

客户端流程

简单NIO示例

AIO简介

主要特点

文件I/O和网络I/O： AIO不仅适用于网络编程，还适用于文件I/O。它可以用于处理文件读写、网络套接字通信等多种I/O操作。

主要组件和类

总结

I/O

I/O（Input/Output）是计算机科学和编程中的一个重要概念，指的是计算机程序与外部世界（通常是硬件设备、文件系统、网络、用户等）进行数据交换和通信的过程。I/O 可以包括从外部设备读取数据（输入）和将数据写入外部设备（输出）两个方面。以下是关于I/O的一些重要概念和说明：

输入（Input）： 输入是指程序从外部获取数据的过程。这可以是从键盘、鼠标、传感器、文件、网络连接等获取数据。例如，用户通过键盘输入文本，程序从网络接收传感器数据，都属于输入操作。
输出（Output）： 输出是指程序向外部设备发送数据的过程。这可以是将数据写入文件、向屏幕显示信息、通过网络传输数据等。例如，程序将计算结果写入文件，向用户显示图形界面，都属于输出操作。
流（Stream）： 流是一种抽象概念，用于表示数据在程序和外部设备之间的有序、连续的传输。I/O流通常分为输入流和输出流。输入流用于从外部设备读取数据，输出流用于将数据发送到外部设备。
阻塞和非阻塞（Blocking and Non-blocking）： 阻塞I/O是指当程序进行I/O操作时，如果没有数据可用或没有完成写入，程序将被阻塞，无法执行其他任务。非阻塞I/O允许程序在等待数据就绪时继续执行其他任务。
同步和异步（Synchronous and Asynchronous）： 同步I/O是指程序在发起I/O操作后等待操作完成，然后继续执行后续任务。异步I/O允许程序发起I/O操作后继续执行其他任务，当操作完成时，程序会收到通知。
文件I/O： 文件I/O是指程序与文件系统进行交互的操作，包括读取文件内容、写入文件、创建文件、删除文件等。文件I/O通常用于数据的持久化和存储。
网络I/O： 网络I/O是指程序与其他计算机通过网络进行通信的操作，包括发送和接收数据、建立连接、断开连接等。网络I/O通常用于构建分布式应用程序。

I/O模型

不同的I/O模型使用不同的方法来处理输入和输出操作，具有不同的特性和适用场景。以下是常见的I/O模型：

阻塞I/O模型（Blocking I/O）： 在阻塞I/O模型中，当应用程序发起一个I/O请求时，程序会被阻塞，直到操作完成或出错。这意味着程序需要等待，不能执行其他任务，直到I/O操作完成。阻塞I/O模型通常简单易用，但可能导致程序性能下降，特别是在高并发环境下。
非阻塞I/O模型（Non-blocking I/O）： 非阻塞I/O模型允许应用程序在等待I/O操作完成时继续执行其他任务，而不会被阻塞。程序可以通过轮询或回调等方式来检查是否有数据可用或操作已完成。非阻塞I/O模型适用于需要高并发处理的场景，但编程复杂度较高。
多路复用I/O模型（Multiplexing I/O）： 多路复用I/O模型使用了选择器（Selector）来监听多个I/O通道的状态，一旦某个通道准备好进行I/O操作，就会触发相应的事件，应用程序可以响应这些事件。这允许一个线程同时管理多个I/O通道，减少线程开销，提高并发性能。常见的多路复用I/O模型包括select、poll和epoll。
信号驱动I/O模型（Signal-driven I/O）： 在信号驱动I/O模型中，应用程序通过注册信号处理程序来处理I/O事件。当I/O操作完成时，操作系统会向应用程序发送信号，应用程序可以捕获信号并处理事件。这种模型通常用于Unix系统。
异步I/O模型（Asynchronous I/O）： 异步I/O模型允许应用程序发起I/O操作后继续执行其他任务，同时注册回调函数来处理I/O操作的完成。操作系统在I/O操作完成后调用回调函数通知应用程序。这种模型适用于需要高度并发和异步操作的场景。

BIO示例

BIO的简单流程

创建BIOServer

public class BIOServer {
    ServerSocket socketServer; // 服务端网络IO的封装对象
    // 构造服务器
    public BIOServer(int port) {
        try {
            socketServer = new ServerSocket(port);
            System.out.println("BIOServer start,Port :" + port);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // 端口有了，需要监听连接
    public void listen() {
        try {
            // 调用accept()阻塞等待客户端连接
            Socket client = socketServer.accept();
            System.out.println("communication port:" + client.getPort());

            InputStream inputStream = client.getInputStream(); // 客户端连接传递的信息流
            BufferedReader clientIn = new BufferedReader(new InputStreamReader(inputStream)); // 通过流读取客户端输入
            PrintWriter serverOut = new PrintWriter(client.getOutputStream()); // 基于socket构造服务端输出对象用于和client发送消息

            // 服务端通过控制台输入向客户端通信发送消息，模拟TCP的全双工通信
            BufferedReader systemIn = new BufferedReader(new InputStreamReader(System.in)); // 通过流读取控制台输入
            String line = systemIn.readLine();
            while (!line.equals("bye")) { // 持续连接
                serverOut.println(line); // 向客户端输出消息
                serverOut.flush(); // 手动将缓冲区中的数据强制刷新到输出流中，以确保数据被立即写入底层的输出流
                System.out.println("receive client msg:" + clientIn.readLine());
                System.out.println("server send msg:" + line);
                line = systemIn.readLine();
            }
            inputStream.close();
            clientIn.close();
            serverOut.close();
            client.close();
            socketServer.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws IOException {
        // 启动服务端，开启监听
        new BIOServer(8080).listen();
    }
}

启动运行Server，创建Client连接Server

public class BIOClient {
    public static void main(String[] args){
        try {
            // 客户端网络IO的封装对象
            Socket socket = new Socket("localhost", 8080);

            // 控制台输入客户端输出信息
            BufferedReader systemIn = new BufferedReader(new InputStreamReader(System.in));
            // 在当前socket连接上写入数据
            PrintWriter clientOut = new PrintWriter(socket.getOutputStream(), true);
            // 获取服务端通过socket发送的输入流
            BufferedReader serverIn = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            // 模拟Tcp全双工通信
            String line = systemIn.readLine();
            while (!line.equals("bye")) { // 持续连接
                clientOut.println(line); // 向服务端输出消息
                clientOut.flush(); // 手动将缓冲区中的数据强制刷新到输出流中，以确保数据被立即写入底层的输出流
                System.out.println("receive server msg:" + serverIn.readLine());
                System.out.println("client send msg:" + line);
                line = systemIn.readLine();
            }
            systemIn.close();
            serverIn.close();
            socket.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

BIO中的阻塞是指读写数据都是单向的，因此是同步阻塞。

BIO与NIO比较

BIO	NIO
阻塞	非阻塞
面向流	面向缓冲区
每个客户端连接都需要一个独立线程处理	选择器，允许一个线程管理多个通道
连接少负载低，适合传统的同步阻塞式服务	高并发性和响应性，适合聊天应用，实时游戏等
编程模型简单，高并发需要管理大量线程	编程模型复杂，具备更好的性能和并发处理能力

NIO的三大核心

NIO有三大核心部分: Channel(通道)，Buffer(缓冲区)，Selector(选择器)

通道（Channel）： 通道是NIO中的基本概念，代表着数据源和数据目标，可以是文件、网络连接、管道等。通道提供了一个可供程序读写数据的抽象接口，不同类型的通道可以用于不同的I/O操作。常见的通道类型包括：
- FileChannel：用于文件I/O操作。
- SocketChannel：用于套接字网络连接。
- ServerSocketChannel：用于服务器套接字。
- DatagramChannel：用于UDP协议数据报的通道。
缓冲区（Buffer）： 缓冲区是NIO中用于存储数据的容器，通常是一个数组。缓冲区用于在通道和应用程序之间传输数据，可以读取数据到缓冲区或将数据从缓冲区写入通道。缓冲区提供了对数据的有序访问，常见的缓冲区类型包括：
- ByteBuffer：用于存储字节数据。
- CharBuffer：用于存储字符数据。
- ShortBuffer、IntBuffer、LongBuffer：用于存储整数数据。
- FloatBuffer、DoubleBuffer：用于存储浮点数数据。
选择器（Selector）： 选择器是NIO中用于多路复用的关键组件，它允许一个线程管理多个通道。通过选择器，可以监视多个通道上的事件，如可读、可写等，并在事件就绪时唤醒相关的线程来处理。选择器的使用使得可以高效地处理多个并发连接，提高了程序的性能和响应性。

NIO核心之缓冲区

缓冲区（Buffer）一个用于特定基本数据类型的容器，所有缓冲区都是 Buffer 抽象类的子类。

Buffer就像一个数组，可以保存多个相同类型的数据。

Buffer常用子类：

ByteBuffer
CharBuffer
ShortBuffer
IntBuffer
LongBuffer
FloatBuffer
DoubleBuffer

创建一个容量为capacity 的 XxxBuffer 对象：

static XxxBuffer allocate(int capacity)

Buffer常用API

Buffer clear()：清空缓冲区，重置位置和限制。
Buffer flip()：切换缓冲区的读写模式，将位置设置为0，并将限制设置为当前位置。
Buffer rewind()：将位置设置为0，保留限制的值，用于重读缓冲区中的数据。
Buffer mark()：在当前位置设置标记。
Buffer reset()：将位置重置为先前设置的标记位置。
boolean hasRemaining()：检查是否还有剩余可读数据。
int remaining()：返回剩余可读数据的数量。
int position()：获取当前位置。
Buffer position(int n)：将设置缓冲区的当前位置为 n.
int limit()：获取限制值的位置。
Buffer limit(int newLimit)：设置限制值的位置为n，并返回修改后的Buffer对象。
int capacity()：返回 Buffer 的 capacity 大小

Buffer中的重要概念

容量 (capacity) ：作为一个内存块，Buffer具有一定的固定大小，也称为"容量"，缓冲区容量不能为负，并且创建后不能更改。
限制 (limit)：表示缓冲区中可以操作数据的大小（limit 后数据不能进行读写）。缓冲区的限制不能为负，并且不能大于其容量。写入模式，限制等于 buffer的容量。读取模式下，limit等于写入的数据量。
位置 (position)：下一个要读取或写入的数据的索引。缓冲区的位置不能为负，并且不能大于其限制

位置、限制、容量遵守以下规律： 0 <= position <= limit <= capacity

1.init初始化容量为10的缓冲区

2.FileChannel.read(buffer)将一个大小为4byte的文件读入缓冲区(put操作)

3.flip()切换读写模式

4.buffer.get()读取缓冲区数据

5.clear()清空缓冲区，方便下次put操作

使用Buffer读写数据一般遵循以下四个步骤：

写入数据到Buffer
调用flip()方法，转换为读取模式
从Buffer中读取数据
调用buffer.clear()方法或者buffer.compact()方法清除缓冲区

NIO核心之通道

Channel 类似于传统的“流”。只不过 Channel 本身不能直接访问数据，Channel 只能与 Buffer 进行交互。

FileChannel 类

获取通道的一种方式是对支持通道的对象调用getChannel() 方法。支持通道的类如下

FileInputStream
FileOutputStream
RandomAccessFile
DatagramSocket
Socket
ServerSocket
获取通道的其他方式是使用 Files 类的静态方法 newByteChannel() 获取字节通道。或者通过通道的静态方法 open() 打开并返回指定通道

FileChannel常用方法

int read(ByteBuffer dst) ：从Channel 到中读取数据到 ByteBuffer
long read(ByteBuffer[] dsts) ： 将Channel中的数据“分散”到 ByteBuffer[]
int write(ByteBuffer src) ：将 ByteBuffer中的数据写入到 Channel
long write(ByteBuffer[] srcs) ：将 ByteBuffer[] 到中的数据“聚集”到 Channel
long position() ：返回此通道的文件位置
FileChannel position(long p) ：设置此通道的文件位置
long size() ：返回此通道的文件的当前大小
FileChannel truncate(long s) ：将此通道的文件截取为给定大小
void force(boolean metaData) ：强制将所有对此通道的文件更新写入到存储设备中

NIO核心之选择器

概述

选择器（Selector)是SelectableChannle对象的多路复用器，Selector可以同时监控多个SelectableChannel的IO状况，也就是说，利用Selector可使一个单独的线程管理多个Channel。Selector是非阻塞IO的核心。

Java 的 NIO，用非阻塞的 IO 方式。可以用一个线程，处理多个的客户端连接，就会使用到 Selector(选择器)
Selector 能够检测多个注册的通道上是否有事件发生(注意:多个 Channel 以事件的方式可以注册到同一个(Selector)，如果有事件发生，便获取事件然后针对每个事件进行相应的处理。这样就可以只用一个单线程去管
理多个通道，也就是管理多个连接和请求。
只有在连接/通道真正有读写事件发生时，才会进行读写，就大大地减少了系统开销，并且不必为每个连接都创建一个线程，不用去维护多个线程
避免了多线程之间的上下文切换导致的开销

应用

创建 Selector ：通过调用 Selector.open() 方法创建一个 Selector。

Selector selector = Selector.open();

向选择器注册通道：SelectableChannel.register(Selector sel, int ops)

//1. 获取通道
ServerSocketChannel ssChannel = ServerSocketChannel.open();
//2. 切换非阻塞模式
ssChannel.configureBlocking(false);
//3. 绑定连接
ssChannel.bind(new InetSocketAddress(port));
//4. 获取选择器
Selector selector = Selector.open();
//5. 将通道注册到选择器上, 并且指定“监听接收事件”
ssChannel.register(selector, SelectionKey.OP_ACCEPT);

当调用 register(Selector sel, int ops) 将通道注册选择器时，选择器对通道的监听事件，需要通过第二个参数 ops 指定。可以监听的事件类型（用可使用 SelectionKey 的四个常量表示）：

读 : SelectionKey.OP_READ （1）
写 : SelectionKey.OP_WRITE （4）
连接 : SelectionKey.OP_CONNECT （8）
接收 : SelectionKey.OP_ACCEPT （16）

NIO非阻塞原理分析

Selector可以实现一个 I/O 线程可以并发处理 N 个客户端连接和读写操作，这从根本上解决了传统同步阻塞 I/O 一连接一线程模型，架构的性能、弹性伸缩能力和可靠性都得到了极大的提升。

服务端流程

1.三大核心配置

// 创建轮询器
Selector selector = Selector.open(); 
// 创建serverSocket通道
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open(); 
// 设置缓冲区
ByteBuffer buffer = ByteBuffer.allocate(1024);

2.阻塞模式设置，绑定端口

// 绑定端口地址
serverSocketChannel.bind(new InetSocketAddress(8088)); 
// NIO为了兼容BIO，默认是阻塞式
serverSocketChannel.configureBlocking(false);

3.注册通道并且指定“监听接收事件”

serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

4.轮询的方式获取选择器上已经“准备就绪”的事件

public void listen() {
        try {
            while (selector.select() > 0) {
                Set<SelectionKey> selectedKeys = selector.selectedKeys(); // 获取已经准备好的事件
                Iterator<SelectionKey> keyIterator = selectedKeys.iterator(); // 迭代器轮询
                while (keyIterator.hasNext()) { // 同步体现在这里，因为每次只能拿一个key，每次只能处理一种状态
                    SelectionKey key = keyIterator.next();
                    process(key); // 处理key代表的业务：就绪、读、写等等
                    keyIterator.remove();
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

public void process(SelectionKey key) throws IOException {
        if(key.isAcceptable()) {
            // 接受客户端连接请求
            SocketChannel socketChannel = serverSocketChannel.accept();
            socketChannel.configureBlocking(false);

            // 将客户端socketChannel注册到selector，并监听读事件
            socketChannel.register(selector, SelectionKey.OP_READ);
        } else if (key.isReadable()) {
            // key.channel 从多路复用器中拿到客户端的引用
            SocketChannel socketChannel = (SocketChannel) key.channel();
            int len = socketChannel.read(buffer);
            if (len > 0) {
                buffer.flip(); // 读写切换，这里是切换为读模式
                String content = new String(buffer.array(), 0, len);
                System.out.println("get receive content:" + content);
                buffer.clear();
            }
        }
    }

客户端流程

1.获取通道

SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress("localhost",8088));

2.切换为非阻塞模式

socketChannel.configureBlocking(false);

3.分配缓冲区大小

ByteBuffer buffer = ByteBuffer.allocate(1024);

4.发送数据给绑定的服务端

            Scanner scanner = new Scanner(System.in);
            while (true){
                System.out.print("请输入:");
                String msg = scanner.nextLine();
                buffer.put(msg.getBytes());
                buffer.flip();
                socketChannel.write(buffer);
                buffer.clear();
            }

简单NIO示例

服务端

public class NIOServer {
    // NIO的三核心：通道(channel)、缓冲区(buffer)、轮询器(selector)

    private int port;
    private Selector selector;
    private ByteBuffer buffer = ByteBuffer.allocate(1024); // 分配大小为1024byte的缓冲区
    private ServerSocketChannel serverSocketChannel;

    public NIOServer(int port) {
        try {
            this.port = port;
            selector = Selector.open(); // 创建轮询器
            serverSocketChannel = ServerSocketChannel.open(); // 创建serverSocket通道
            serverSocketChannel.bind(new InetSocketAddress(this.port)); // 绑定端口地址
            serverSocketChannel.configureBlocking(false); // NIO为了兼容BIO，默认是阻塞式
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT); // 将serverSocket通道注册到selector上监听连接事件
            System.out.println("Server start on port ：" + this.port);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void listen() {
        try {
            while (selector.select() > 0) {
                Set<SelectionKey> selectedKeys = selector.selectedKeys(); // 获取已经准备好的事件
                Iterator<SelectionKey> keyIterator = selectedKeys.iterator(); // 迭代器轮询
                while (keyIterator.hasNext()) { // 同步体现在这里，因为每次只能拿一个key，每次只能处理一种状态
                    SelectionKey key = keyIterator.next();
                    process(key); // 处理key代表的业务：就绪、读、写等等
                    keyIterator.remove();
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    public void process(SelectionKey key) throws IOException {
        if(key.isAcceptable()) {
            // 接受客户端连接请求
            SocketChannel socketChannel = serverSocketChannel.accept();
            socketChannel.configureBlocking(false);

            // 将客户端socketChannel注册到selector，并监听读事件
            socketChannel.register(selector, SelectionKey.OP_READ);
        } else if (key.isReadable()) {
            // key.channel 从多路复用器中拿到客户端的引用
            SocketChannel socketChannel = (SocketChannel) key.channel();
            int len = socketChannel.read(buffer);
            if (len > 0) {
                buffer.flip(); // 读写切换，这里是切换为读模式
                String content = new String(buffer.array(), 0, len);
                System.out.println("get receive content:" + content);
                buffer.clear();
            }
        }
    }

    public static void main(String[] args) {
        new NIOServer(8088).listen();
    }
}

客户端

public class NIOClient {
    public static void main(String[] args) {
        try {
            SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress("localhost",8088));
            socketChannel.configureBlocking(false);
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            Scanner scanner = new Scanner(System.in);
            while (true){
                System.out.print("请输入:");
                String msg = scanner.nextLine();
                buffer.put(msg.getBytes());
                buffer.flip();
                socketChannel.write(buffer);
                buffer.clear();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

AIO简介

AIO（Asynchronous I/O）是Java NIO（New I/O）库的一部分，用于实现异步非阻塞I/O操作。与传统的BIO（Blocking I/O）和NIO（Non-blocking I/O）不同，AIO通过回调机制在I/O操作完成时通知应用程序，而不需要应用程序一直轮询等待数据准备或操作完成。

以下是Java AIO的简介和主要特点：

主要特点

异步操作： AIO提供了异步I/O操作，允许应用程序在发起I/O请求后继续执行其他任务，而不必等待I/O操作完成。
回调机制： AIO使用回调机制通知应用程序有关I/O操作的完成情况。应用程序可以注册回调函数，当I/O操作完成时，操作系统会调用回调函数来处理数据。
适用于高并发： AIO适用于高并发环境，可以处理大量的并发连接和I/O操作，而无需为每个连接创建一个线程。
文件I/O和网络I/O： AIO不仅适用于网络编程，还适用于文件I/O。它可以用于处理文件读写、网络套接字通信等多种I/O操作。

主要组件和类

AsynchronousChannel： AIO的通道类，包括AsynchronousFileChannel（文件I/O）和AsynchronousSocketChannel（套接字网络通信）等。这些通道支持异步I/O操作。
CompletionHandler： 用于注册回调函数的接口，包括CompletedHandler<V, A>，其中V表示I/O操作的结果类型，A表示附加的上下文对象。应用程序通过实现这个接口来处理I/O操作完成的通知。
AsynchronousServerSocketChannel： 用于实现异步的服务器套接字通信，可以监听并接受连接请求。

总结

BIO（Blocking I/O）、NIO（Non-blocking I/O）和AIO（Asynchronous I/O）各自适用于不同的场景，以下是它们的主要适用场景：

BIO（Blocking I/O）：
- 适用场景： BIO适用于连接数量较少且每个连接的负载较低的场景。它在每个连接上都创建一个独立的线程来处理I/O操作，因此不适合高并发的环境。
- 示例应用： 传统的同步阻塞式服务器，如Web服务器的早期版本，FTP服务器等。
NIO（Non-blocking I/O）：
- 适用场景： NIO适用于高并发和高吞吐量的网络应用，其中需要处理大量并发连接。它通过复用少量线程来管理多个通道，减少了线程开销，提高了性能和资源利用率。
- 示例应用： 高性能的网络服务器、聊天应用、实时游戏服务器、代理服务器等。
AIO（Asynchronous I/O）：
- 适用场景： AIO适用于需要高并发、异步处理的应用，尤其是在处理文件I/O或需要等待时间较长的I/O操作时。它通过回调机制在I/O操作完成时通知应用程序，允许应用程序继续执行其他任务，不必等待I/O操作完成。
- 示例应用： 文件上传/下载服务、邮件服务器、高吞吐量的文件传输应用等。