基于Quarkus 3 Mutiny框架的Base64流式解码代码正确性咨询

我正在学习Quarkus 3的Mutiny框架，需要处理一段较大的Base64字符串，将其解码为字符串以便后续保存到文件中。我希望每次读取4字节（符合Base64编码特性），并以最优多线程方式实现。以下是我编写的代码，请问这段代码是否合理？是否有优化空间？

@Inject
ManagedExecutor managedExecutor;

List<byte[]> resultDecoded = readBase64InStreamingFashionAndWriteTo(resultEncoded).collect().asList().await().indefinitely();
// 合并字节数组为字符串
String resultMerge = resultDecoded.stream().map(b -> new String(b, StandardCharsets.UTF_8)).reduce("", (resultString, partialStr) -> resultString + partialStr);
logger.info("Result Decoded: " + resultMerge);

public Multi<byte[]> readBase64InStreamingFashionAndWriteTo(String base64Str) throws IOException {
    logger.info("start thread read " + Thread.currentThread());
    AtomicInteger i = new AtomicInteger(1);
    return Uni.createFrom().item(base64Str)
           .emitOn(managedExecutor) // 工作线程池
            .log()
           .onItem().transform(String::getBytes)
           .onItem().transform(ByteArrayInputStream::new)
           .onItem().transformToMulti(this::readChunkAsync)// 异步处理
            .emitOn(managedExecutor)
            .onItem().transform( str -> {
               if (i.getAndAdd(1) <= 2) logger.info(Thread.currentThread() + "**read partial string**: " + str);
               return  str;})
           .onItem().transform(DatatypeConverter::parseBase64Binary);
}

private Multi<String> readChunkAsync(ByteArrayInputStream in) {
    return Multi.createFrom().emitter( em -> {  // 发射读取到的Base64字符串片段
              try {
                  AtomicInteger i = new AtomicInteger(1);
                  byte[] buf = new byte[4];// 每次读4字节
                  int nread = -1;
                  while ((nread = in.read(buf, 0, buf.length)) != -1) {
                      if (i.getAndAdd(1) <= 2)  logger.info(Thread.currentThread() + "**emiting**: ");
                      em.emit(new String(buf, StandardCharsets.UTF_8));
                  }
              }catch(Exception ex){
                  // 空异常处理
              } finally {
                  em.complete();
              }
            }
    );
}

代码存在的问题 #

• 分块读取逻辑错误：当ByteArrayInputStream最后一次读取不足4字节时，buf中会残留之前的字节内容，导致生成的Base64片段包含无效字符，最终解码失败。比如原Base64长度不是4的倍数时，最后一段会出现错误。
• 多线程使用冗余：readChunkAsync中的Emitter是同步循环发射数据，没有真正实现异步处理；两次调用emitOn(managedExecutor)会增加不必要的线程切换开销，反而降低性能。
• 结果合并效率极低：使用reduce拼接字符串时，每次都会创建新的String对象，处理大内容时会产生大量临时对象，触发频繁GC，严重影响性能。
• 异常处理缺失：readChunkAsync中的catch块为空，会吞掉所有异常，导致问题无法被及时发现和排查。
• 调试代码冗余：用于日志的AtomicInteger属于调试代码，生产环境无实际意义，应该移除。

优化建议与改进代码 #

核心优化点 #

1. 修正分块读取逻辑：根据实际读取的字节数截取有效部分，避免无效字符。
2. 简化线程模型：仅在需要异步执行的步骤使用emitOn，减少线程切换。
3. 流式写入文件：直接通过Multi的订阅方法写入文件，无需收集成List再合并，大幅降低内存占用。
4. 使用高效解码器：替换DatatypeConverter为Java自带的Base64.getDecoder()，更高效且类型安全。
5. 完善异常处理：在Emitter中传递异常，方便上层捕获处理。
6. 减少不必要的类型转换：直接处理字节块，避免String与byte[]的频繁转换。

优化后的代码示例 #

@Inject
ManagedExecutor managedExecutor;

// 直接流式处理并写入文件，无需收集到List
public void processBase64AndWriteToFile(String base64Str, Path outputPath) {
    readBase64InStreamingFashion(base64Str)
        .emitOn(managedExecutor) // 解码操作在工作线程池执行
        .subscribe()
        .with(
            decodedBytes -> Files.write(outputPath, decodedBytes, StandardOpenOption.CREATE, StandardOpenOption.APPEND),
            throwable -> logger.error("处理Base64失败", throwable)
        );
}

public Multi<byte[]> readBase64InStreamingFashion(String base64Str) {
    return Uni.createFrom().item(base64Str)
            .onItem().transformToMulti(this::splitBase64IntoChunks)
            .onItem().map(Base64.getDecoder()::decode);
}

private Multi<String> splitBase64IntoChunks(String base64Str) {
    return Multi.createFrom().emitter(em -> {
        try {
            int length = base64Str.length();
            int index = 0;
            // 按4字节（Base64的一个编码块）拆分字符串
            while (index < length) {
                int end = Math.min(index + 4, length);
                String chunk = base64Str.substring(index, end);
                em.emit(chunk);
                index = end;
            }
            em.complete();
        } catch (Exception ex) {
            em.fail(ex); // 传递异常
        }
    });
}

额外优化说明 #

• 如果Base64字符串极大，建议直接从输入流读取（比如HTTP请求体、文件流），而不是先加载为String，进一步降低内存占用。
• 写入文件时可以使用BufferedOutputStream包装，减少IO操作次数，提升写入性能。

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