如何用Mutiny异步非阻塞读取大文件中的Protobuf消息？

你的代码存在两个关键问题：

1. 跨块消息截断：Vertx的AsyncFile.toMulti()是分块读取文件的，单个Protobuf分隔消息可能跨多个块，原代码把每个块单独解析，会导致跨块的消息无法被正确识别。
2. 不必要的列表收集：每个块解析后先收集到列表再转成Multi，虽然功能上可行，但可以优化为解析出一个消息就立刻发射，减少内存占用。

要实现真正的异步非阻塞、逐元素读取，核心是维护一个累积缓冲区，用来保存未处理的字节（比如某个块末尾不完整的消息），每次新块到来时合并缓冲区，循环解析所有能完整读取的消息，剩余的不完整字节留到下一次处理。

完整代码实现 #

@Inject
io.vertx.mutiny.core.Vertx vertx;

// 用来累积未处理的字节，处理跨块消息
static class ByteAccumulator {
    private final ByteArrayOutputStream buffer = new ByteArrayOutputStream();

    void add(byte[] bytes) throws IOException {
        buffer.write(bytes);
    }

    byte[] getCurrentBytes() {
        return buffer.toByteArray();
    }

    // 保留从offset开始的剩余字节，用于下一次解析
    void retainRemaining(int offset) {
        byte[] remaining = Arrays.copyOfRange(buffer.toByteArray(), offset, buffer.size());
        buffer.reset();
        try {
            buffer.write(remaining);
        } catch (IOException e) {
            throw new RuntimeException("Failed to retain remaining bytes", e);
        }
    }

    void clear() {
        buffer.reset();
    }
}

Multi<ProtoMessage> readProtoMessages() {
    Path path = Path.of("my/file");
    ByteAccumulator accumulator = new ByteAccumulator();

    return vertx.fileSystem()
            .open(path.toString(), new OpenOptions().setRead(true))
            // 将AsyncFile转为按块读取的Multi<Buffer>，可自定义块大小（示例为8KB）
            .onItem().transformToMulti(asyncFile -> asyncFile.toMulti(8192))
            // 提取Buffer中的字节数组
            .onItem().transform(Buffer::getBytes)
            // 处理每个块的字节，解析出所有完整的Protobuf消息
            .onItem().transformToMultiAndConcatenate(bytes -> {
                try {
                    accumulator.add(bytes);
                    List<ProtoMessage> parsedMessages = new ArrayList<>();
                    ByteArrayInputStream inputStream = new ByteArrayInputStream(accumulator.getCurrentBytes());
                    int totalProcessedBytes = 0;

                    ProtoMessage message;
                    // 循环解析所有能完整读取的消息
                    while ((message = ProtoMessage.parseDelimitedFrom(inputStream)) != null) {
                        parsedMessages.add(message);
                        // 计算已处理的字节数：总字节数 - 输入流剩余字节数
                        totalProcessedBytes = accumulator.getCurrentBytes().length - inputStream.available();
                    }

                    // 保留未处理的剩余字节（不完整的消息）
                    accumulator.retainRemaining(totalProcessedBytes);

                    // 直接发射当前块解析出的所有消息，无需提前收集全量列表
                    return Multi.createFrom().iterable(parsedMessages);
                } catch (IOException e) {
                    // 解析失败时发射错误信号
                    return Multi.createFrom().failure(new RuntimeException("Failed to parse Protobuf message", e));
                }
            })
            // 流终止时清理缓冲区
            .onTermination().invoke(accumulator::clear);
}

关键细节说明 #

• ByteAccumulator：负责累积未处理的字节，确保跨块的消息不会被截断。每次解析完成后，只保留未处理的剩余字节，避免内存无限增长。
• 异步分块读取：asyncFile.toMulti(8192)指定了块大小（可根据文件大小和消息大小调整），Vertx会异步分块读取文件，不会一次性加载整个大文件到内存，避免阻塞事件循环。
• 逐元素发射：每解析出一个完整的Protobuf消息，就会立刻通过Multi发射出去，不需要先把所有消息收集到列表中，适合处理超大文件。

关于你担心的open()方法阻塞问题：Vertx的fileSystem.open()是异步非阻塞操作，它不会一次性把整个文件读入Buffer，只是打开文件句柄，后续的读取是通过toMulti()异步分块进行的，所以不会阻塞事件循环。

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