深入浅出分布式服务网络通信 BIO NIO AIO 详解附代码

BIO、NIO、AIO

同步和异步

同步和异步是指应用程序与内核交互的方式：

同步 (Synchronize)：用户进程触发IO操作后，必须等待IO操作完成或轮询检查IO操作是否就绪。例如：传统的BIO模式中，当调用read()方法时，必须等到数据真正从内核空间拷贝到用户空间后才能返回。
异步 (Asychronize)：当异步调用发出后，调用者不会立即得到结果。被调用者通过状态改变通知（如NIO的Selector机制）或回调函数处理（如AIO的CompletionHandler）通知调用者。

阻塞与非阻塞

阻塞和非阻塞是针对进程访问数据时的行为方式：

阻塞：读写操作会一直等待，直到IO操作完成。在数据未准备好时，调用线程会被挂起。优点是编程模型简单，缺点是资源利用率低。
非阻塞：读写操作立即返回状态值，不会等待。优点是提高系统吞吐量，缺点是需要轮询检查状态，增加CPU负担。

组合应用场景

同步阻塞(BIO)：典型实现是Java传统Socket，适用场景是连接数少且固定的架构。
同步非阻塞(NIO)：典型实现是Java NIO的Selector机制，适用场景是高并发连接但连接时间短的架构（如聊天服务器）。
异步非阻塞(AIO)：典型实现是Java AIO，适用场景是高并发连接且连接时间长的架构（如文件服务器）。

注意：不存在”异步阻塞”的组合，因为异步本身就意味着不阻塞调用线程。

BIO (Blocking I/O)

基本介绍

BIO（Blocking I/O）即同步阻塞I/O模型，其中的”B”代表Blocking（阻塞）。这是Java1.4版本之前唯一的I/O模型选择，也是传统网络编程中最基础的I/O处理方式。

工作原理

服务器实现采用”一个连接一个线程”的模式。当客户端发起连接请求时：

服务端必须为每个新连接分配一个独立的线程进行处理
该线程会全程负责该连接的所有I/O操作
在I/O操作（如read/write）期间，线程会一直阻塞等待，直到操作完成

性能特点

资源开销大：每个连接都需要独立的线程，线程创建和上下文切换成本高
简单易用：编程模型直观，易于理解和实现
效率低下：当连接数增加时，线程数量线性增长，系统资源很快耗尽

适用场景

客户端连接数量较少且固定的应用
对延迟不敏感的服务
需要快速开发的原型系统
教学示例和简单演示程序

服务端代码

package icu.wzk.io;

import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;

public class WzkIOServer {

    public static void main(String[] args) throws Exception {
        ServerSocket serverSocket = new ServerSocket();
        serverSocket.bind(new InetSocketAddress("127.0.0.1", 8889));
        while (true) {
            Socket socket = serverSocket.accept();
            new Thread(() -> {
                try {
                    byte[] bytes = new byte[1024];
                    int len = socket.getInputStream().read(bytes);
                    System.out.println(new String(bytes, 0, len));

                    socket.getOutputStream().write(bytes, 0, len);
                    socket.getOutputStream().flush();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }

}

客户端代码

package icu.wzk.io;

import java.net.Socket;

public class WzkIOClient {

    public static void main(String[] args) throws Exception {
        Socket socket = new Socket("127.0.0.1", 8889);
        socket.getOutputStream().write("hello".getBytes());
        socket.getOutputStream().flush();
        System.out.println("============ send ===========");

        byte[] bytes = new byte[1024];
        int len = socket.getInputStream().read(bytes);
        System.out.println("server: " + new String(bytes, 0, len));
        socket.close();
    }

}

NIO (New I/O)

基本介绍

NIO（Non-blocking I/O 或 New I/O）是 Java 从 JDK1.4 版本开始提供的一套全新 I/O API，用于替代传统的阻塞式 I/O。与传统的 I/O 相比，NIO 提供了更高效、更灵活的 I/O 处理方式。

NIO 的核心特点：

同步非阻塞：线程不会因为等待 I/O 操作完成而被阻塞
基于通道和缓冲区的 I/O 方式
使用单线程管理多个连接的能力

服务器实现模式：NIO 采用”一个请求一个通道”的模式，所有客户端连接请求都会注册到多路复用器（Selector）上。多路复用器不断轮询这些通道，当某个通道有 I/O 请求时，才会启动一个线程进行处理。

通道 (Channels)

通道是 NIO 引入的最重要的抽象概念，它代表了一个开放的数据连接，可以用于读取和写入数据。与流不同，通道是双向的，可以同时用于读写操作。

主要特点：

数据可以从 Channel 读取到 Buffer 中
数据可以从 Buffer 写入到 Channel 中
支持异步非阻塞操作
提供多种实现类型（如 FileChannel、SocketChannel 等）

缓冲区 (Buffers)

缓冲区是 NIO 用于存储数据的容器，所有数据都是通过缓冲区进行传输的。

缓冲区关键操作：

写入数据到 Buffer
调用 flip() 切换为读模式
从 Buffer 中读取数据
调用 clear() 或 compact() 清空缓冲区

选择器 (Selectors)

选择器是 NIO 实现多路复用的关键组件，允许单线程处理多个通道。

工作原理：

将通道注册到选择器上
选择器不断轮询已注册的通道
当通道上有感兴趣的事件发生时，选择器会通知应用程序
应用程序处理这些事件

主要事件类型：

SelectionKey.OP_CONNECT（连接就绪）
SelectionKey.OP_ACCEPT（接受就绪）
SelectionKey.OP_READ（读就绪）
SelectionKey.OP_WRITE（写就绪）

服务端代码

package icu.wzk.nio;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.Scanner;

public class WzkNIOServer extends Thread {

    private Selector selector;

    private final ByteBuffer readBuffer = ByteBuffer.allocate(1024);

    private final ByteBuffer writeBuffer = ByteBuffer.allocate(1024);

    public WzkNIOServer(int port) {
        init(port);
    }

    private void init(int port) {
        try {
            System.out.println("Server 启动");
            this.selector = Selector.open();
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.configureBlocking(false);
            serverSocketChannel.bind(new InetSocketAddress(port));
            serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);
            System.out.println("Server 启动完成");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void run() {
        while (true) {
            try {
                this.selector.select();
                Iterator<SelectionKey> keys = this.selector.selectedKeys().iterator();
                while (keys.hasNext()) {
                    SelectionKey key = keys.next();
                    keys.remove();
                    if (key.isValid()) {
                        try {
                            if (key.isAcceptable()) {
                                accept(key);
                            }
                            if (key.isReadable()) {
                                read(key);
                            }
                            if (key.isWritable()) {
                                write(key);
                            }
                        } catch (CancelledKeyException e) {
                            key.cancel();
                        }
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }

        }
    }

    private void accept(SelectionKey key) {
        try {
            ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
            SocketChannel socketChannel = serverSocketChannel.accept();
            socketChannel.configureBlocking(false);
            socketChannel.register(this.selector, SelectionKey.OP_READ);
            System.out.println("客户端连接成功: " + socketChannel.getRemoteAddress());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void read(SelectionKey key) {
        try {
            this.readBuffer.clear();
            SocketChannel socketChannel = (SocketChannel) key.channel();
            int readLen = socketChannel.read(this.readBuffer);
            if (readLen == -1) {
                key.channel().close();
                return;
            }

            this.readBuffer.flip();
            byte[] bytes = new byte[readBuffer.remaining()];
            readBuffer.get(bytes);
            System.out.println("从客户端接收到了: " + socketChannel.getRemoteAddress() + " " + new String(bytes, StandardCharsets.UTF_8));

            socketChannel.register(this.selector, SelectionKey.OP_WRITE);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void write(SelectionKey key) {
        try {
            this.writeBuffer.clear();
            SocketChannel socketChannel = (SocketChannel) key.channel();
            Scanner scanner = new Scanner(System.in);
            System.out.println("发送数据到客户端, 输入后回车");
            String line = scanner.nextLine();
            this.writeBuffer.put(line.getBytes(StandardCharsets.UTF_8));
            writeBuffer.flip();
            socketChannel.write(writeBuffer);

            socketChannel.register(this.selector, SelectionKey.OP_READ);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        new Thread(new WzkNIOServer(8889)).start();
    }

}

客户端代码

package icu.wzk.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;


public class WzkNIOClient {
    public static void main(String[] args) {
        InetSocketAddress inetSocketAddress = new InetSocketAddress("127.0.0.1", 8889);
        SocketChannel socketChannel = null;
        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
        try {
            socketChannel = SocketChannel.open();
            socketChannel.connect(inetSocketAddress);
            Scanner sc = new Scanner(System.in);
            while (true) {
                System.out.println("客户端给服务器发消息");
                String line = sc.nextLine();
                if ("exit".equals(line)) {
                    break;
                }
                byteBuffer.put(line.getBytes(StandardCharsets.UTF_8));
                byteBuffer.flip();
                socketChannel.write(byteBuffer);
                byteBuffer.clear();

                int len = socketChannel.read(byteBuffer);
                if (len == -1) {
                    socketChannel.close();
                    break;
                }
                byteBuffer.flip();

                byte[] bytes = new byte[byteBuffer.remaining()];
                byteBuffer.get(bytes);
                System.out.println("服务器数据: " + new String(bytes, StandardCharsets.UTF_8));
                byteBuffer.clear();
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (null != socketChannel) {
                try {
                    socketChannel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

}