基于Mutiny的Base64编码优化：多线程流式处理大文件方案问询

我需要读取一个大型Base64字符串并解码为String后保存到文件（未来可能直接操作文件）。希望每次读取4字节（符合Base64编码特性），并采用最优多线程方式实现。请问当前代码是否合理？该如何优化？

当前代码 #

Optional<String> result1 = convertFileToBase64InStreamingFashion2("c:/temp/prova.txt").collect().asList()
    .await().indefinitely()
     .stream().reduce((acc, el) -> acc.concat(el));
      logger.info(result1.get());
    
    public Multi<String> convertFileToBase64InStreamingFashion2(String path ) throws IOException {
        logger.info("start thread " + Thread.currentThread());
        AtomicInteger i = new AtomicInteger(1);
        return vertx.fileSystem().open(path, new OpenOptions().setRead(true))
                .log()
                .onItem().invoke(
                        x-> x.setReadBufferSize(3*18))
                .onItem().transformToMulti(AsyncFile::toMulti)
                .log()
                .emitOn(managedExecutor)
                .onItem().transform(b ->
                     new Base64().encodeAsString(b.getBytes()))
                .log();

    }

日志输出 #

2024-04-17 22:57:01,638 INFO  [org.acm.GreetingService] (Quarkus Main Thread) start thread Thread[Quarkus Main Thread,5,build group]
2024-04-17 22:57:01,639 INFO  [io.qua.mut.run.MutinyInfrastructure] (Quarkus Main Thread) Uni.AsyncResultUni.0 | onSubscribe()
2024-04-17 22:57:01,640 INFO  [io.qua.mut.run.MutinyInfrastructure] (Quarkus Main Thread) Multi.UniOnItemTransformToMulti.0 | onSubscribe()
2024-04-17 22:57:01,640 INFO  [io.qua.mut.run.MutinyInfrastructure] (Quarkus Main Thread) Multi.MultiMapOp.0 | onSubscribe()
2024-04-17 22:57:01,640 INFO  [io.qua.mut.run.MutinyInfrastructure] (Quarkus Main Thread) Multi.MultiMapOp.0 | request(9223372036854775807)
2024-04-17 22:57:01,640 INFO  [io.qua.mut.run.MutinyInfrastructure] (Quarkus Main Thread) Multi.UniOnItemTransformToMulti.0 | request(16)
2024-04-17 22:57:01,641 INFO  [io.qua.mut.run.MutinyInfrastructure] (vert.x-eventloop-thread-4) Uni.AsyncResultUni.0 | onItem(io.vertx.core.file.impl.AsyncFileImpl@45a0f69c)
2024-04-17 22:57:01,642 INFO  [io.qua.mut.run.MutinyInfrastructure] (vert.x-eventloop-thread-4) Multi.UniOnItemTransformToMulti.0 | onItem(#du)
2024-04-17 22:57:01,642 INFO  [io.qua.mut.run.MutinyInfrastructure] (executor-thread-1) Multi.MultiMapOp.0 | onItem(I2R1)

..

2024-04-17 22:57:10,150 INFO  [io.qua.mut.run.MutinyInfrastructure] (executor-thread-2) Multi.UniOnItemTransformToMulti.0 | request(16)
2024-04-17 22:57:10,152 INFO  [io.qua.mut.run.MutinyInfrastructure] (vert.x-eventloop-thread-4) Multi.UniOnItemTransformToMulti.0 | onCompletion()
2024-04-17 22:57:10,152 INFO  [io.qua.mut.run.MutinyInfrastructure] (executor-thread-2) Multi.MultiMapOp.0 | onCompletion()

当前代码的不合理之处 #

1. 块大小与Base64特性不匹配：设置的readBufferSize=3*18既不符合Base64解码（每4字节对应3字节原始数据）也不符合编码（每3字节对应4字节编码）的对齐要求，会导致块边界的Base64片段不完整，处理时可能报错或出现乱码。
2. 字符串拼接效率极低：先收集所有片段到列表再用String.concat()拼接，每次拼接都会创建新字符串，大型文件下会产生大量临时对象，严重消耗内存和CPU。
3. 多线程利用不充分：emitOn(managedExecutor)仅切换了编码操作的线程，但没有结合批量处理减少线程切换开销，整体流处理的并行效率未最大化。
4. Base64实例重复创建：每次transform都新建Base64实例，而Base64实例是线程安全的，重复创建属于不必要的性能浪费。
5. 业务逻辑与需求不符：当前代码是将文件内容编码为Base64，但你的需求是读取Base64字符串解码后保存，逻辑完全反向。

优化方案 #

1. 对齐Base64处理的块大小 #

• 解码场景：设置readBufferSize=4，每次读取完整的4字节Base64单元（对应3字节原始数据），最后一块不足4字节时单独处理补位的=字符。
• 编码场景（若后续需要）：设置readBufferSize=3，保证每次读取的原始数据能完整编码为4字节Base64。

2. 高效处理结果：直接写入文件而非内存拼接 #

避免将整个文件加载到内存，利用Vert.x异步IO直接在流中写入解码后的数据，同时用批量处理减少IO次数：

private final Base64.Decoder base64Decoder = Base64.getDecoder();

public void decodeBase64File(String inputPath, String outputPath) {
    vertx.fileSystem().open(inputPath, new OpenOptions().setRead(true))
            .onItem().invoke(asyncFile -> asyncFile.setReadBufferSize(4))
            .onItem().transformToMulti(AsyncFile::toMulti)
            .emitOn(managedExecutor) // 解码操作放线程池，避免阻塞事件循环
            .onItem().transform(buffer -> {
                String base64Chunk = buffer.toString();
                return base64Decoder.decode(base64Chunk);
            })
            .batch(128) // 批量处理，减少IO调用次数
            .onItem().transformToUni(batch -> {
                Buffer mergedBuffer = Buffer.buffer();
                batch.forEach(mergedBuffer::appendBytes);
                return vertx.fileSystem().writeFile(outputPath, mergedBuffer, new WriteOptions().setAppend(true));
            })
            .subscribe().with(
                    success -> logger.info("解码完成"),
                    error -> logger.error("解码失败", error)
            );
}

3. 优化多线程与流处理 #

• 使用emitOn(managedExecutor)将CPU密集型的解码操作委托给线程池，避免阻塞Vert.x事件循环线程。
• 用batch()批量处理数据，减少线程切换和IO操作的开销，提升整体处理效率。
• 复用Base64解码器实例，避免重复创建对象。

4. 避免阻塞调用 #

在Quarkus这类反应式框架中，尽量不要使用await().indefinitely()阻塞线程，改用非阻塞的subscribe()处理结果，保证应用的反应式特性。

额外注意事项 #

• 处理超大型文件时，全程使用异步流操作，避免内存溢出。
• 若Base64文件包含换行符或空白字符，需先过滤掉再解码，否则会导致解码失败。

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