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

Java NIO 非阻塞模式下的网络编程技巧

2021-09-055.6k 阅读

Java NIO 非阻塞模式概述

在传统的 Java 网络编程中，基于 java.net 包的 Socket 和 ServerSocket 实现的是阻塞式 I/O。这意味着当执行诸如 read() 或 accept() 等操作时，线程会被阻塞，直到数据可读或有新的连接到来。这种方式在处理大量并发连接时效率较低，因为每个连接都需要一个独立的线程来处理 I/O 操作，会消耗大量的系统资源。

Java NIO（New I/O）引入了非阻塞 I/O 的概念，它允许在单个线程中管理多个通道（Channel）。非阻塞 I/O 的核心思想是，当请求的 I/O 操作无法立即完成时，不会阻塞线程，而是返回一个特定的状态值，让程序可以继续执行其他任务，稍后再检查操作是否完成。

核心组件

  1. 通道（Channel）：通道是 NIO 中用于执行 I/O 操作的实体，它类似于传统 I/O 中的流，但提供了更灵活的操作方式。通道可以是双向的，既可以读也可以写，而传统的流通常是单向的（输入流或输出流）。常见的通道类型有 SocketChannel（用于 TCP 套接字通信）、ServerSocketChannel（用于监听新的 TCP 连接）、DatagramChannel（用于 UDP 通信）等。
  2. 缓冲区（Buffer）：缓冲区是 NIO 中用于存储数据的地方。所有的数据都要先写入缓冲区，然后再从缓冲区读取。缓冲区本质上是一个数组，但它提供了更丰富的操作方法，如 position、limit 和 capacity 等属性来管理数据的读写。常见的缓冲区类型有 ByteBuffer、CharBuffer、IntBuffer 等。
  3. 选择器（Selector）：选择器是 NIO 非阻塞模式的关键组件，它允许一个线程监视多个通道的 I/O 事件。通过将通道注册到选择器上，并指定要监听的事件类型（如读事件、写事件等），选择器可以轮询这些通道，当有感兴趣的事件发生时，通知程序进行相应的处理。

非阻塞模式下的 TCP 网络编程

创建 ServerSocketChannel

在非阻塞模式下创建一个 TCP 服务器，首先需要创建 ServerSocketChannel 并将其设置为非阻塞模式。

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.ServerSocketChannel;

public class NonBlockingServer {
    public static void main(String[] args) {
        try {
            // 创建 ServerSocketChannel
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
            // 绑定端口
            serverSocketChannel.socket().bind(new InetSocketAddress(8888));
            // 设置为非阻塞模式
            serverSocketChannel.configureBlocking(false);
            System.out.println("Server started, listening on port 8888");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

在上述代码中，通过 ServerSocketChannel.open() 创建一个 ServerSocketChannel 实例，然后使用 socket().bind() 方法绑定到指定的端口 8888，最后通过 configureBlocking(false) 将其设置为非阻塞模式。

注册选择器

接下来，需要创建一个选择器，并将 ServerSocketChannel 注册到选择器上，监听连接事件（SelectionKey.OP_ACCEPT）。

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;

public class NonBlockingServer {
    public static void main(String[] args) {
        try {
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.socket().bind(new InetSocketAddress(8888));
            serverSocketChannel.configureBlocking(false);

            // 创建选择器
            Selector selector = Selector.open();
            // 将 ServerSocketChannel 注册到选择器上，监听连接事件
            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 server = (ServerSocketChannel) key.channel();
                        SocketChannel client = server.accept();
                        client.configureBlocking(false);
                        System.out.println("New client connected: " + client);
                        // 将新连接的 SocketChannel 注册到选择器上，监听读事件
                        client.register(selector, SelectionKey.OP_READ);
                    } else if (key.isReadable()) {
                        // 处理读事件
                        SocketChannel client = (SocketChannel) key.channel();
                        ByteBuffer buffer = ByteBuffer.allocate(1024);
                        int bytesRead = client.read(buffer);
                        if (bytesRead > 0) {
                            buffer.flip();
                            byte[] data = new byte[buffer.limit()];
                            buffer.get(data);
                            String message = new String(data);
                            System.out.println("Received from client: " + message);
                        }
                    }

                    keyIterator.remove();
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

在上述代码中，通过 Selector.open() 创建一个选择器实例，然后使用 serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT) 将 ServerSocketChannel 注册到选择器上，监听连接事件。在 while (true) 循环中，通过 selector.select() 阻塞等待事件发生，当有事件发生时，获取已就绪的事件集合，并根据事件类型进行相应的处理。

处理新连接

当有新的连接到来时（key.isAcceptable()），通过 server.accept() 接受新的连接，并将新连接的 SocketChannel 设置为非阻塞模式，然后将其注册到选择器上，监听读事件。

if (key.isAcceptable()) {
    ServerSocketChannel server = (ServerSocketChannel) key.channel();
    SocketChannel client = server.accept();
    client.configureBlocking(false);
    System.out.println("New client connected: " + client);
    client.register(selector, SelectionKey.OP_READ);
}

处理读事件

当有可读事件发生时（key.isReadable()），从 SocketChannel 中读取数据。

if (key.isReadable()) {
    SocketChannel client = (SocketChannel) key.channel();
    ByteBuffer buffer = ByteBuffer.allocate(1024);
    int bytesRead = client.read(buffer);
    if (bytesRead > 0) {
        buffer.flip();
        byte[] data = new byte[buffer.limit()];
        buffer.get(data);
        String message = new String(data);
        System.out.println("Received from client: " + message);
    }
}

在上述代码中，首先创建一个 ByteBuffer 用于存储读取的数据，然后通过 client.read(buffer) 读取数据。如果读取到的数据长度大于 0，则将缓冲区翻转（buffer.flip()），并将数据读取到字节数组中，最后转换为字符串输出。

非阻塞模式下的 UDP 网络编程

创建 DatagramChannel

在非阻塞模式下进行 UDP 通信，需要创建 DatagramChannel 并将其设置为非阻塞模式。

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;

public class NonBlockingUdpServer {
    public static void main(String[] args) {
        try {
            // 创建 DatagramChannel
            DatagramChannel datagramChannel = DatagramChannel.open();
            // 绑定端口
            datagramChannel.bind(new InetSocketAddress(9999));
            // 设置为非阻塞模式
            datagramChannel.configureBlocking(false);
            System.out.println("UDP Server started, listening on port 9999");

            ByteBuffer buffer = ByteBuffer.allocate(1024);
            InetSocketAddress clientAddress = (InetSocketAddress) datagramChannel.receive(buffer);
            if (clientAddress != null) {
                buffer.flip();
                byte[] data = new byte[buffer.limit()];
                buffer.get(data);
                String message = new String(data);
                System.out.println("Received from client: " + message);

                // 发送响应
                String response = "Message received!";
                buffer.clear();
                buffer.put(response.getBytes());
                buffer.flip();
                datagramChannel.send(buffer, clientAddress);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

在上述代码中，通过 DatagramChannel.open() 创建一个 DatagramChannel 实例，然后使用 bind() 方法绑定到指定的端口 9999，最后通过 configureBlocking(false) 将其设置为非阻塞模式。

接收和发送数据

在非阻塞模式下，通过 datagramChannel.receive(buffer) 接收数据，如果接收到数据，则处理数据并发送响应。

ByteBuffer buffer = ByteBuffer.allocate(1024);
InetSocketAddress clientAddress = (InetSocketAddress) datagramChannel.receive(buffer);
if (clientAddress != null) {
    buffer.flip();
    byte[] data = new byte[buffer.limit()];
    buffer.get(data);
    String message = new String(data);
    System.out.println("Received from client: " + message);

    // 发送响应
    String response = "Message received!";
    buffer.clear();
    buffer.put(response.getBytes());
    buffer.flip();
    datagramChannel.send(buffer, clientAddress);
}

在上述代码中，首先创建一个 ByteBuffer 用于存储接收的数据，然后通过 datagramChannel.receive(buffer) 接收数据。如果接收到数据（clientAddress != null），则将缓冲区翻转，读取数据并转换为字符串输出。接着，准备响应数据，将其写入缓冲区并发送给客户端。

缓冲区的深入理解

缓冲区的状态属性

  1. capacity：缓冲区的容量，即缓冲区能够容纳的数据元素的最大数量。一旦缓冲区创建，其容量就固定不变。
  2. position：当前的读写位置，读操作从 position 位置开始读取数据，写操作从 position 位置开始写入数据，每次读写操作后，position 会相应地增加。
  3. limit：表示缓冲区中可以读写的数据的界限。在写模式下，limit 通常等于 capacity；在读模式下，limit 表示上次写操作后实际写入的数据量。

缓冲区的操作方法

  1. flip()：将缓冲区从写模式切换到读模式。在写模式下，position 记录的是已写入的数据位置，调用 flip() 后，limit 被设置为当前的 position，position 被重置为 0，这样就可以从缓冲区的起始位置开始读取已写入的数据。
  2. rewind()：将 position 重置为 0，limit 保持不变。通常用于重新读取缓冲区中的数据，而不需要切换模式。
  3. clear()：将 position 重置为 0，limit 设置为 capacity。该方法主要用于准备缓冲区进行下一次写入操作，但不会清空缓冲区中的数据。

直接缓冲区和非直接缓冲区

  1. 直接缓冲区：直接缓冲区是通过调用 ByteBuffer.allocateDirect(capacity) 创建的缓冲区。直接缓冲区在堆外内存中分配空间，这样可以减少数据在堆内存和直接内存之间的拷贝，提高 I/O 性能。直接缓冲区适用于频繁的 I/O 操作，但创建和销毁的开销较大。
  2. 非直接缓冲区：非直接缓冲区是通过调用 ByteBuffer.allocate(capacity) 创建的缓冲区，它在堆内存中分配空间。非直接缓冲区的创建和销毁开销较小，但在进行 I/O 操作时，需要将数据从堆内存拷贝到直接内存，性能相对较低。

选择器的高级应用

监听多种事件

选择器可以同时监听多种 I/O 事件，如读事件（SelectionKey.OP_READ）、写事件（SelectionKey.OP_WRITE）、连接事件（SelectionKey.OP_CONNECT）和接收事件（SelectionKey.OP_ACCEPT）。通过在注册通道时指定多个事件类型，可以在一个线程中处理多种不同类型的 I/O 操作。

SocketChannel client = SocketChannel.open();
client.configureBlocking(false);
client.connect(new InetSocketAddress("localhost", 8888));

Selector selector = Selector.open();
client.register(selector, SelectionKey.OP_CONNECT | SelectionKey.OP_READ | SelectionKey.OP_WRITE);

在上述代码中，SocketChannel 注册到选择器上，同时监听连接、读和写事件。

动态调整监听事件

可以在运行时动态调整通道注册的监听事件。例如，当某个通道暂时没有数据可读，但需要写入数据时，可以将其监听事件从 OP_READ 调整为 OP_WRITE。

SelectionKey key = clientChannel.keyFor(selector);
key.interestOps(SelectionKey.OP_WRITE);

在上述代码中，通过 key.interestOps(SelectionKey.OP_WRITE) 将通道的监听事件调整为只监听写事件。

处理多个选择器

在一些复杂的应用场景中，可能需要使用多个选择器来管理不同类型的通道或提高系统的并发性能。例如，可以将处理 I/O 操作的选择器和处理定时任务的选择器分开，避免相互干扰。

Selector ioSelector = Selector.open();
Selector timerSelector = Selector.open();

// 将 I/O 相关的通道注册到 ioSelector
ServerSocketChannel serverChannel = ServerSocketChannel.open();
serverChannel.configureBlocking(false);
serverChannel.register(ioSelector, SelectionKey.OP_ACCEPT);

// 将定时任务相关的通道（假设存在）注册到 timerSelector
// 这里只是示例，实际可能需要更复杂的实现
// TimerChannel timerChannel = TimerChannel.open();
// timerChannel.configureBlocking(false);
// timerChannel.register(timerSelector, SelectionKey.OP_READ);

// 启动两个线程分别处理两个选择器
Thread ioThread = new Thread(() -> {
    while (true) {
        try {
            int readyChannels = ioSelector.select();
            if (readyChannels == 0) continue;
            Set<SelectionKey> selectedKeys = ioSelector.selectedKeys();
            // 处理 I/O 事件
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
});

Thread timerThread = new Thread(() -> {
    while (true) {
        try {
            int readyChannels = timerSelector.select();
            if (readyChannels == 0) continue;
            Set<SelectionKey> selectedKeys = timerSelector.selectedKeys();
            // 处理定时任务事件
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
});

ioThread.start();
timerThread.start();

在上述代码中，创建了两个选择器 ioSelector 和 timerSelector，并分别启动两个线程来处理这两个选择器上的事件。

性能优化与注意事项

缓冲区大小的选择

缓冲区大小的选择对性能有重要影响。如果缓冲区过小，可能导致频繁的读写操作，增加系统开销；如果缓冲区过大，会浪费内存空间，并且可能影响数据传输的及时性。在实际应用中，需要根据具体的业务场景和数据量来合理选择缓冲区大小。一般来说，对于网络 I/O，常见的缓冲区大小为 8192 字节（8KB）。

避免不必要的拷贝

在使用 NIO 时，应尽量避免不必要的数据拷贝。例如，使用直接缓冲区可以减少数据在堆内存和直接内存之间的拷贝次数。此外，在处理数据时，可以尽量复用已有的缓冲区，而不是频繁地创建新的缓冲区。

选择器的轮询效率

选择器的轮询效率直接影响系统的性能。在高并发场景下，选择器的轮询可能成为性能瓶颈。可以通过合理设置选择器的轮询超时时间（selector.select(timeout)）来优化性能。如果设置的超时时间过短，选择器会频繁地进行轮询，增加系统开销；如果设置的超时时间过长，可能导致事件处理不及时。一般来说，需要根据实际的业务需求和系统负载来调整超时时间。

异常处理

在非阻塞网络编程中，异常处理非常重要。例如，在读取或写入数据时，可能会遇到 IOException，需要合理地处理这些异常，避免程序崩溃。在处理连接事件时，也可能会遇到连接超时、连接拒绝等异常，需要进行相应的处理，如关闭通道、重新连接等。

try {
    int bytesRead = clientChannel.read(buffer);
    if (bytesRead == -1) {
        // 客户端关闭连接
        clientChannel.close();
    }
} catch (IOException e) {
    e.printStackTrace();
    // 处理异常，如关闭通道等
    try {
        clientChannel.close();
    } catch (IOException ex) {
        ex.printStackTrace();
    }
}

在上述代码中，当读取数据时如果返回 -1，表示客户端关闭连接，此时关闭通道。如果发生 IOException，则打印异常信息并关闭通道。

内存管理

由于 NIO 涉及到直接内存的使用，需要注意内存管理。直接内存的分配和释放不受 Java 垃圾回收机制的管理，因此需要手动释放直接内存。可以通过调用 ByteBuffer 的 cleaner() 方法获取 Cleaner 对象，然后在适当的时候调用 Cleaner 的 clean() 方法释放直接内存。但需要注意的是，Cleaner 是基于引用队列实现的，可能存在一定的延迟。

ByteBuffer directBuffer = ByteBuffer.allocateDirect(1024);
// 使用 directBuffer
// 手动释放直接内存
sun.misc.Cleaner cleaner = ((DirectBuffer) directBuffer).cleaner();
if (cleaner != null) {
    cleaner.clean();
}

在上述代码中，通过 ((DirectBuffer) directBuffer).cleaner() 获取 Cleaner 对象，然后调用 clean() 方法释放直接内存。

与阻塞式网络编程的对比

资源消耗

阻塞式网络编程为每个连接创建一个独立的线程来处理 I/O 操作，当并发连接数较多时，会消耗大量的系统资源，如线程栈空间、CPU 上下文切换开销等。而 NIO 非阻塞模式通过选择器在单个线程中管理多个通道，大大减少了线程的数量，降低了资源消耗。

性能表现

在高并发场景下，阻塞式网络编程由于线程的阻塞和上下文切换，性能会受到较大影响。NIO 非阻塞模式能够更有效地利用系统资源，避免线程的无效等待，提高系统的吞吐量和响应速度。但在低并发场景下，由于 NIO 的实现相对复杂，引入了选择器等组件，其性能可能不如阻塞式网络编程。

编程复杂度

阻塞式网络编程的代码结构相对简单，逻辑清晰，易于理解和维护。而 NIO 非阻塞模式的编程复杂度较高，需要深入理解通道、缓冲区、选择器等概念，并且在处理复杂业务逻辑时，代码的可读性和可维护性可能会受到一定影响。

实际应用场景

高性能服务器

在开发高性能的网络服务器，如 Web 服务器、游戏服务器等场景中，NIO 非阻塞模式可以充分发挥其优势，处理大量的并发连接，提高服务器的性能和吞吐量。

分布式系统

在分布式系统中，节点之间的通信需要高效的网络编程。NIO 非阻塞模式可以满足分布式系统对高并发、低延迟通信的需求，例如在分布式缓存系统、分布式计算框架中广泛应用。

物联网（IoT）

物联网设备通常需要与服务器进行大量的并发通信。NIO 非阻塞模式可以有效地管理这些连接，实现设备与服务器之间的高效数据传输，适用于智能家居、工业物联网等领域。

通过以上对 Java NIO 非阻塞模式下网络编程技巧的详细介绍，包括核心组件、TCP 和 UDP 编程、缓冲区和选择器的深入理解、性能优化以及与阻塞式编程的对比等方面，希望能帮助开发者更好地掌握和应用 NIO 非阻塞模式进行高效的网络编程。在实际应用中，需要根据具体的业务需求和系统场景，合理选择和优化相关的技术和策略，以达到最佳的性能和用户体验。