MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

Java NIO 非阻塞 I/O 提升系统并发的实践

2022-02-125.9k 阅读

Java NIO 基础概述

Java NIO（New I/O）是从 Java 1.4 开始引入的一套新的 I/O 体系，它提供了与传统 I/O 不同的操作方式。传统的 Java I/O 是面向流（Stream - oriented）的，操作是以字节或字符为单位顺序进行的，并且通常是阻塞式的。而 NIO 是面向缓冲区（Buffer - oriented）和通道（Channel - based）的，支持非阻塞 I/O 操作，这使得在处理高并发 I/O 场景时效率更高。

缓冲区（Buffer）

缓冲区是 NIO 中用于存储数据的地方。在 NIO 中，所有的数据都是通过缓冲区来处理的。缓冲区本质上是一块内存区域，它提供了对数据的结构化访问以及维护读写位置等信息。

Java NIO 中有多种类型的缓冲区，如 ByteBuffer、CharBuffer、ShortBuffer、IntBuffer、LongBuffer、FloatBuffer 和 DoubleBuffer，它们都继承自抽象类 java.nio.Buffer。

以 ByteBuffer 为例，创建一个 ByteBuffer 的方式如下：

// 创建一个容量为 1024 的 ByteBuffer
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);

allocate 方法用于分配指定容量大小的缓冲区。ByteBuffer 有几个重要的属性：

  • 容量（Capacity）：缓冲区能够容纳的数据元素的最大数量。一旦缓冲区被创建，其容量就固定了。
  • 位置（Position）：下一个要读取或写入的数据元素的索引。当从缓冲区写入数据时，位置会随着数据的写入而增加；当从缓冲区读取数据时，位置同样会增加。
  • 限制（Limit）：缓冲区中可以读取或写入的最后一个数据元素的索引加 1。在写入模式下，限制通常等于容量；在读取模式下，限制会被设置为写入模式下的位置，表示可以读取的有效数据范围。

例如，当我们向 ByteBuffer 写入数据后，需要切换到读取模式，可以使用 flip() 方法：

byte[] data = "Hello, NIO!".getBytes();
byteBuffer.put(data);
byteBuffer.flip();
byte[] result = new byte[byteBuffer.limit()];
byteBuffer.get(result);
System.out.println(new String(result));

flip() 方法将限制设置为当前位置，然后将位置重置为 0，这样就准备好从缓冲区读取数据了。

通道（Channel）

通道是 NIO 中用于与 I/O 设备（如文件、套接字等）进行交互的对象。与传统 I/O 中的流不同，通道是双向的，可以同时进行读和写操作，而流通常是单向的（要么是输入流，要么是输出流）。

NIO 中有几种主要类型的通道，如 FileChannel 用于文件 I/O，SocketChannel 和 ServerSocketChannel 用于网络套接字 I/O 等。

以 FileChannel 为例，读取文件内容的代码如下：

try (FileInputStream fileInputStream = new FileInputStream("example.txt");
     FileChannel fileChannel = fileInputStream.getChannel()) {
    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
    int bytesRead = fileChannel.read(byteBuffer);
    while (bytesRead != -1) {
        byteBuffer.flip();
        byte[] data = new byte[byteBuffer.limit()];
        byteBuffer.get(data);
        System.out.println(new String(data));
        byteBuffer.clear();
        bytesRead = fileChannel.read(byteBuffer);
    }
} catch (IOException e) {
    e.printStackTrace();
}

这里通过 FileInputStream 获取 FileChannel，然后使用 FileChannel 的 read 方法将数据读取到 ByteBuffer 中。每次读取后，需要对 ByteBuffer 进行 flip 操作来准备读取数据，读取完毕后再使用 clear 方法重置缓冲区以便下一次读取。

非阻塞 I/O 原理

在传统的阻塞式 I/O 模型中，当一个线程执行 I/O 操作（如读取网络数据或文件数据）时，该线程会被阻塞，直到 I/O 操作完成。这意味着在 I/O 操作进行的过程中，线程无法执行其他任务，从而浪费了 CPU 资源。

而在非阻塞 I/O 模型中，当线程发起 I/O 操作时，线程不会被阻塞，而是立即返回。如果 I/O 操作尚未完成，线程可以继续执行其他任务，然后在适当的时候再次检查 I/O 操作的状态，看是否已经完成。

非阻塞 I/O 与 Selector

在 Java NIO 中，实现非阻塞 I/O 的关键组件是 Selector。Selector 允许一个线程监控多个通道的 I/O 事件，例如连接建立、数据可读、数据可写等。通过使用 Selector，一个线程可以管理多个通道，从而显著提高系统的并发处理能力。

要使用 Selector，首先需要将通道设置为非阻塞模式。以 SocketChannel 为例：

SocketChannel socketChannel = SocketChannel.open();
socketChannel.configureBlocking(false);

然后，将通道注册到 Selector 上，并指定要监听的事件类型。事件类型主要有以下几种：

  • OP_READ：表示通道有数据可读。
  • OP_WRITE：表示通道可以写入数据。
  • OP_CONNECT：表示套接字连接操作已完成（仅适用于 SocketChannel）。
  • OP_ACCEPT：表示服务器套接字通道已准备好接受新的连接（仅适用于 ServerSocketChannel）。

注册通道到 Selector 的代码如下：

Selector selector = Selector.open();
socketChannel.register(selector, SelectionKey.OP_CONNECT);

register 方法返回一个 SelectionKey 对象，它代表了通道和 Selector 之间的注册关系。通过 SelectionKey，可以获取注册的通道以及所监听的事件等信息。

Selector 的核心方法是 select()，该方法会阻塞，直到注册的通道中有至少一个通道发生了所监听的事件。当 select() 方法返回时，我们可以通过 selectedKeys() 方法获取发生事件的 SelectionKey 集合，然后遍历该集合处理相应的事件。

下面是一个简单的示例，展示了如何使用 Selector 监听多个 SocketChannel 的可读事件：

Selector selector = Selector.open();
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.bind(new InetSocketAddress(8080));
serverSocketChannel.configureBlocking(false);
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

while (true) {
    int readyChannels = selector.select();
    if (readyChannels == 0) continue;

    Set<SelectionKey> selectedKeys = selector.selectedKeys();
    Iterator<SelectionKey> keyIterator = selectedKeys.iterator();

    while (keyIterator.hasNext()) {
        SelectionKey key = keyIterator.next();

        if (key.isAcceptable()) {
            ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel();
            SocketChannel clientChannel = serverChannel.accept();
            clientChannel.configureBlocking(false);
            clientChannel.register(selector, SelectionKey.OP_READ);
        } else if (key.isReadable()) {
            SocketChannel clientChannel = (SocketChannel) key.channel();
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            int bytesRead = clientChannel.read(byteBuffer);
            if (bytesRead > 0) {
                byteBuffer.flip();
                byte[] data = new byte[byteBuffer.limit()];
                byteBuffer.get(data);
                System.out.println("Received: " + new String(data));
            }
        }

        keyIterator.remove();
    }
}

在这个示例中，ServerSocketChannel 注册了 OP_ACCEPT 事件，当有新的客户端连接时，接受连接并将新的 SocketChannel 设置为非阻塞模式，然后注册 OP_READ 事件。在 while 循环中，通过 selector.select() 等待事件发生，当有事件发生时，根据事件类型进行相应的处理。处理完事件后，需要从 selectedKeys 集合中移除已处理的 SelectionKey，以避免重复处理。

非阻塞 I/O 在提升系统并发中的应用场景

网络服务器

在网络服务器开发中，非阻塞 I/O 可以显著提升服务器的并发处理能力。传统的阻塞式 I/O 服务器在处理大量并发连接时，每个连接都需要一个独立的线程来处理 I/O 操作，这会导致线程数量过多，消耗大量的系统资源，并且线程上下文切换也会带来额外的开销。

使用非阻塞 I/O 和 Selector，一个线程可以管理多个连接的 I/O 操作，只有当连接有数据可读或可写时才进行处理，大大减少了线程的数量和上下文切换的开销。例如，一个简单的 HTTP 服务器可以使用 NIO 来实现高效的并发处理：

Selector selector = Selector.open();
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.bind(new InetSocketAddress(8080));
serverSocketChannel.configureBlocking(false);
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

while (true) {
    int readyChannels = selector.select();
    if (readyChannels == 0) continue;

    Set<SelectionKey> selectedKeys = selector.selectedKeys();
    Iterator<SelectionKey> keyIterator = selectedKeys.iterator();

    while (keyIterator.hasNext()) {
        SelectionKey key = keyIterator.next();

        if (key.isAcceptable()) {
            ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel();
            SocketChannel clientChannel = serverChannel.accept();
            clientChannel.configureBlocking(false);
            clientChannel.register(selector, SelectionKey.OP_READ);
        } else if (key.isReadable()) {
            SocketChannel clientChannel = (SocketChannel) key.channel();
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            StringBuilder request = new StringBuilder();
            int bytesRead;
            while ((bytesRead = clientChannel.read(byteBuffer)) != -1) {
                byteBuffer.flip();
                request.append(new String(byteBuffer.array(), 0, bytesRead));
                byteBuffer.clear();
            }
            // 处理 HTTP 请求
            String response = handleHttpRequest(request.toString());
            ByteBuffer responseBuffer = ByteBuffer.wrap(response.getBytes());
            clientChannel.write(responseBuffer);
            clientChannel.close();
        }

        keyIterator.remove();
    }
}

在这个示例中，服务器监听新的连接（OP_ACCEPT 事件），当有新连接时接受连接并注册 OP_READ 事件。当有数据可读时，读取 HTTP 请求数据，处理请求并返回响应。这种方式可以高效地处理大量并发的 HTTP 请求，而不需要为每个请求创建单独的线程。

分布式系统中的数据传输

在分布式系统中，节点之间需要进行大量的数据传输。例如，一个分布式文件系统中，客户端与各个存储节点之间需要频繁地进行文件的读取和写入操作。如果使用阻塞式 I/O，每个数据传输操作都会阻塞线程，导致系统的并发性能低下。

使用非阻塞 I/O，节点可以在等待数据传输完成的同时，继续处理其他任务，如处理其他客户端的请求、进行本地数据的维护等。通过 Selector 管理多个网络通道，可以有效地协调多个数据传输操作，提高系统的整体并发性能。

例如，在一个分布式数据同步系统中，一个节点可能需要与多个其他节点进行数据同步：

Selector selector = Selector.open();
List<SocketChannel> socketChannels = new ArrayList<>();
// 初始化与多个节点的连接
for (InetSocketAddress address : targetAddresses) {
    SocketChannel socketChannel = SocketChannel.open();
    socketChannel.connect(address);
    socketChannel.configureBlocking(false);
    socketChannel.register(selector, SelectionKey.OP_CONNECT);
    socketChannels.add(socketChannel);
}

while (true) {
    int readyChannels = selector.select();
    if (readyChannels == 0) continue;

    Set<SelectionKey> selectedKeys = selector.selectedKeys();
    Iterator<SelectionKey> keyIterator = selectedKeys.iterator();

    while (keyIterator.hasNext()) {
        SelectionKey key = keyIterator.next();

        if (key.isConnectable()) {
            SocketChannel socketChannel = (SocketChannel) key.channel();
            if (socketChannel.isConnectionPending()) {
                socketChannel.finishConnect();
            }
            socketChannel.register(selector, SelectionKey.OP_READ);
        } else if (key.isReadable()) {
            SocketChannel socketChannel = (SocketChannel) key.channel();
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            int bytesRead = socketChannel.read(byteBuffer);
            if (bytesRead > 0) {
                byteBuffer.flip();
                // 处理接收到的数据
                handleReceivedData(byteBuffer);
            }
        }

        keyIterator.remove();
    }
}

在这个示例中，节点初始化与多个目标节点的连接，并将连接通道注册到 Selector 上监听连接建立事件（OP_CONNECT）。连接建立后，注册 OP_READ 事件来读取其他节点发送的数据。这样，节点可以同时与多个其他节点进行数据交互，提高了分布式系统的数据传输并发性能。

非阻塞 I/O 实践中的注意事项

缓冲区管理

在使用非阻塞 I/O 时，缓冲区的管理非常重要。由于数据可能不会一次性全部读取或写入，需要合理地处理缓冲区的容量、位置和限制等属性。

例如，在读取数据时，可能需要多次调用 read 方法才能读取完所有数据。每次读取后，需要根据读取的字节数来调整缓冲区的位置和限制，以确保正确地处理数据。同样，在写入数据时，也需要注意缓冲区的剩余空间，避免数据溢出。

ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
int totalBytesRead = 0;
while (true) {
    int bytesRead = socketChannel.read(byteBuffer);
    if (bytesRead == -1) {
        break;
    }
    totalBytesRead += bytesRead;
    if (byteBuffer.remaining() == 0) {
        // 缓冲区已满，需要扩大缓冲区或处理已读取的数据
        ByteBuffer newBuffer = ByteBuffer.allocate(byteBuffer.capacity() * 2);
        byteBuffer.flip();
        newBuffer.put(byteBuffer);
        byteBuffer = newBuffer;
    }
}

在这个示例中，当缓冲区已满（remaining() == 0）时，创建一个更大的缓冲区来继续读取数据，以确保不会丢失数据。

事件处理顺序

在处理 Selector 监听到的事件时，需要注意事件的处理顺序。例如，在处理 OP_CONNECT 事件时，需要确保连接完全建立后（通过 finishConnect 方法）再注册其他事件（如 OP_READ）。

如果处理顺序不当，可能会导致程序逻辑错误。例如，在连接尚未完全建立时就尝试读取数据，会导致 read 方法返回错误。

if (key.isConnectable()) {
    SocketChannel socketChannel = (SocketChannel) key.channel();
    if (socketChannel.isConnectionPending()) {
        socketChannel.finishConnect();
    }
    socketChannel.register(selector, SelectionKey.OP_READ);
}

这样可以确保在连接建立成功后再注册 OP_READ 事件，以便正确地读取数据。

异常处理

在非阻塞 I/O 操作中，可能会出现各种异常，如连接超时、网络中断等。需要在代码中妥善处理这些异常，以保证系统的稳定性。

例如，在读取或写入数据时，如果发生 IOException，需要关闭相应的通道，并从 Selector 中取消注册，避免无效的事件处理。

try {
    int bytesRead = socketChannel.read(byteBuffer);
    if (bytesRead == -1) {
        socketChannel.close();
        key.cancel();
    }
} catch (IOException e) {
    e.printStackTrace();
    socketChannel.close();
    key.cancel();
}

在这个示例中，当读取数据发生异常或连接关闭（read 返回 -1）时，关闭通道并取消 SelectionKey 的注册，以避免后续的无效操作。

非阻塞 I/O 与其他并发模型的比较

与多线程阻塞式 I/O 模型比较

多线程阻塞式 I/O 模型为每个 I/O 操作创建一个独立的线程。这种模型的优点是编程简单，每个线程专注于一个 I/O 操作，逻辑清晰。然而，随着并发连接数的增加，线程数量也会相应增加，导致系统资源消耗过大。

线程需要占用一定的内存空间（每个线程的栈空间默认大小通常为 1MB 左右），大量线程会消耗大量的内存。此外，线程上下文切换也会带来额外的开销，降低系统的整体性能。

相比之下，非阻塞 I/O 使用 Selector 实现单线程管理多个通道，大大减少了线程数量，降低了内存消耗和上下文切换开销。在高并发场景下，非阻塞 I/O 模型具有更高的性能和可扩展性。

与异步 I/O 模型比较

异步 I/O 模型与非阻塞 I/O 模型有些相似，但也有区别。在异步 I/O 中，应用程序发起 I/O 操作后，不需要主动检查操作状态，系统会在 I/O 操作完成后通过回调函数或事件通知应用程序。

而在非阻塞 I/O 中，虽然线程不会被阻塞，但需要应用程序主动轮询 Selector 来检查 I/O 事件是否发生。异步 I/O 更加适合那些对实时性要求较高，且不需要应用程序立即处理结果的场景。

Java 的 NIO 提供的是非阻塞 I/O 功能，虽然可以通过一些方式模拟异步 I/O（如使用 Future 类来获取 I/O 操作的结果），但与真正的异步 I/O 模型还是有所不同。在实际应用中，需要根据具体的需求和场景选择合适的 I/O 模型。

非阻塞 I/O 的性能优化

合理设置缓冲区大小

缓冲区大小对非阻塞 I/O 的性能有一定影响。如果缓冲区设置得过小，可能会导致频繁的缓冲区扩容操作，增加系统开销。例如，在网络传输中，如果每次读取或写入的数据量较大，过小的缓冲区会导致多次读写操作，降低传输效率。

另一方面，如果缓冲区设置得过大，会浪费内存空间。在实际应用中，需要根据数据的特点和传输需求来合理设置缓冲区大小。对于网络传输，可以根据网络带宽和预期的数据包大小来估算合适的缓冲区大小。

例如，在一个基于 NIO 的文件传输应用中，如果文件以较大的块进行传输，可以设置较大的缓冲区，如 8KB 或 16KB：

ByteBuffer byteBuffer = ByteBuffer.allocate(16384);

这样可以减少缓冲区扩容的次数，提高文件传输的效率。

优化 Selector 操作

Selector 的性能对整个非阻塞 I/O 系统的性能至关重要。首先，减少 Selector 中注册的通道数量可以提高 select 方法的性能。如果某些通道在一段时间内不会有 I/O 操作，可以考虑从 Selector 中取消注册，当需要时再重新注册。

此外，合理设置 select 方法的超时时间也很重要。如果超时时间设置得过短，会导致 select 方法频繁返回，增加系统开销；如果设置得过长，可能会导致 I/O 事件处理不及时。

// 设置 select 方法的超时时间为 100 毫秒
int readyChannels = selector.select(100);

根据实际应用场景，通过实验和调优来确定合适的超时时间，可以提高 Selector 的性能。

使用 DirectByteBuffer

DirectByteBuffer 是一种特殊的 ByteBuffer，它直接分配在堆外内存中。与普通的 ByteBuffer（分配在 Java 堆内）相比，DirectByteBuffer 可以减少数据在堆内和堆外之间的复制，提高 I/O 性能。

在一些对性能要求较高的场景，如网络数据传输和文件 I/O 中，可以考虑使用 DirectByteBuffer：

ByteBuffer byteBuffer = ByteBuffer.allocateDirect(1024);

然而，DirectByteBuffer 的分配和回收比普通 ByteBuffer 更复杂，并且会占用更多的系统内存，所以需要谨慎使用，并根据系统的内存情况进行合理的调整。

总结

Java NIO 的非阻塞 I/O 为提升系统并发性能提供了强大的工具。通过使用缓冲区、通道和 Selector，可以实现高效的 I/O 操作，减少线程数量，降低系统资源消耗。在实际应用中，需要深入理解非阻塞 I/O 的原理，合理处理缓冲区管理、事件处理顺序和异常处理等问题。

与其他并发模型相比，非阻塞 I/O 具有独特的优势，尤其在高并发场景下表现出色。通过性能优化，如合理设置缓冲区大小、优化 Selector 操作和使用 DirectByteBuffer 等，可以进一步提升非阻塞 I/O 的性能。

在未来的开发中，随着系统对并发性能要求的不断提高，Java NIO 的非阻塞 I/O 将在网络服务器、分布式系统等领域发挥更加重要的作用。开发者需要不断掌握和应用这些技术，以构建高效、可扩展的系统。