基于Java NIO从非阻塞SocketChannel读取Protobuf数据的疑问

我正在尝试使用Java NIO开发一个简单的客户端-服务器应用。

Protobuf 定义 #

message ArrRequest {
  int32 n = 1;
  repeated int32 array = 2;
}

message ArrResponse {
  int32 n = 1;
  repeated int32 array = 2;
}

客户端发送代码（阻塞方式） #

try (Socket socket = new Socket("localhost", Constant.PORT)) {
    for (int i = 0; i < numberOfRequestsPerClient; i++) {
        List<Integer> arr = new ArrayList<>();
        Random r = new Random();
        for (int j = 0; j < numberOfArrayElements; j++) {
            arr.add(r.nextInt());
        }
        var request = ArrRequest.newBuilder()
                .setN(numberOfArrayElements)
                .addAllArray(arr)
                .build();
        request.writeDelimitedTo(socket.getOutputStream());
        // 省略后续代码
    }
}

服务器端读取代码片段（非阻塞NIO） #

client.configureBlocking(false); // 关键设置

selector.select();
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> iterator = keys.iterator();
while (iterator.hasNext()) {
    SelectionKey key = iterator.next();
    ClientData data = (ClientData) key.attachment();

    if (key.isReadable()) {
        int bytesRead = data.socket.read(data.buffer); // socket为非阻塞模式
        // 省略后续代码
    }
    // 省略其他事件处理
}

问题 #

我会将数据分批读取到ByteBuffer中，现在我的问题是：何时停止读取？
何时可以将ByteBuffer转换为ArrRequest？具体该如何操作？以及如何获取ArrRequest的字节大小？

解决方案 #

客户端用writeDelimitedTo发送消息时，Protobuf会先写入Varint编码的消息长度，再写入消息本体。服务器端需要对应分两步处理：先解析长度，再读取足够的消息体字节后解析。

核心思路

给每个客户端连接维护一个状态机，分两个阶段处理：

1. 解析Varint格式的消息长度，确定需要读取的消息体总字节数
2. 持续读取数据，直到ByteBuffer中的数据足够容纳完整消息体，再解析为ArrRequest

具体实现步骤

1. 扩展ClientData类，添加状态管理字段：

class ClientData {
    SocketChannel socket;
    ByteBuffer buffer;
    // 状态标识：0=读取消息长度，1=读取消息体
    private int state = 0;
    // 解析出的消息总长度
    private int messageLength = 0;
    // 已读取的长度字节数（Varint最多5字节）
    private int lengthBytesRead = 0;

    // 构造方法等省略
}

2. 在可读事件中处理分阶段读取：

if (key.isReadable()) {
    int bytesRead = data.socket.read(data.buffer);
    if (bytesRead == -1) {
        // 客户端断开，清理资源
        key.cancel();
        data.socket.close();
        iterator.remove();
        return;
    }

    data.buffer.flip(); // 切换为读模式

    while (data.buffer.hasRemaining()) {
        if (data.state == 0) {
            // 阶段1：解析Varint长度
            while (data.buffer.hasRemaining() && data.lengthBytesRead < 5) {
                byte b = data.buffer.get();
                // 拼接Varint数据：低7位有效，左移对应偏移
                data.messageLength |= (b & 0x7F) << (7 * data.lengthBytesRead);
                // 最高位为0表示长度解析完成
                if ((b & 0x80) == 0) {
                    data.state = 1;
                    break;
                }
                data.lengthBytesRead++;
            }
        } else {
            // 阶段2：读取并解析消息体
            if (data.buffer.remaining() >= data.messageLength) {
                // 读取完整消息体字节
                byte[] messageBytes = new byte[data.messageLength];
                data.buffer.get(messageBytes);
                // 解析为ArrRequest
                ArrRequest request = ArrRequest.parseFrom(messageBytes);
                
                // 这里添加你的业务逻辑处理request
                System.out.println("收到请求：n=" + request.getN());

                // 重置状态，准备接收下一条消息
                data.state = 0;
                data.messageLength = 0;
                data.lengthBytesRead = 0;
            } else {
                // 消息体未读完，退出循环等待下一次可读事件
                break;
            }
        }
    }

    data.buffer.compact(); // 将未读完的数据移到缓冲区开头，准备下次写入
    iterator.remove();
}

关键细节说明

• Varint解析：Protobuf的writeDelimitedTo用Varint编码消息长度，这种编码1-5字节即可表示整数，最高位标记是否还有后续字节。
• ByteBuffer操作：flip()切换读写模式，compact()保留未处理的数据，避免丢失分片数据。
• 非阻塞适配：不能直接用ArrRequest.parseDelimitedFrom()，因为它是阻塞式的，非阻塞场景下必须手动分阶段处理。

内容的提问来源于stack exchange，提问作者nutella_eater