FilePart转BufferedReader遇DataBufferUtils.write阻塞问题求解决方案

你的代码出现阻塞的核心原因是PipedInputStream/PipedOutputStream的线程模型冲突：

• DataBufferUtils.write是异步执行的（基于Reactor线程池），而主线程直接同步读取PipedInputStream，当管道缓冲区满时，写操作会阻塞；同时主线程卡在读取步骤，无法消费缓冲区数据，最终形成死锁。

下面提供两种可行的解决方案：

方案一：同步读取（适合阻塞式控制器场景） #

直接将FilePart的内容合并为单个DataBuffer，再转换为InputStream进行同步读取，避免管道流的线程冲突：

@PostMapping("/importData")
public List<BulkImportResponse<ContainerBulkImportObj>> importData(@RequestPart(value = "file", required = true) FilePart file) {
    log.debug("[importData] BEGIN ");
    List<BulkImportResponse<ContainerBulkImportObj>> result = new ArrayList<>();
    try {
        // 合并FilePart的所有数据块为单个DataBuffer
        DataBuffer mergedBuffer = DataBufferUtils.join(file.content()).block();
        if (mergedBuffer != null) {
            // 转换为InputStream并包装为BufferedReader
            try (InputStream inputStream = mergedBuffer.asInputStream();
                 BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream))) {
                result = bulkImportService.importData(bufferedReader);
            } finally {
                // 手动释放DataBuffer，避免内存泄漏
                DataBufferUtils.release(mergedBuffer);
            }
        }
    } catch (Exception ex) {
        log.error("[importData] An error occured during execution:", ex);
    }
    log.debug("[importData] END ");
    return result;
}

方案二：非阻塞异步处理（适合WebFlux原生场景） #

如果你的项目是纯WebFlux架构，建议将控制器改为返回Mono，保持整个链路非阻塞：

@PostMapping("/importData")
public Mono<List<BulkImportResponse<ContainerBulkImportObj>>> importData(@RequestPart(value = "file", required = true) FilePart file) {
    log.debug("[importData] BEGIN ");
    return DataBufferUtils.join(file.content())
            .map(mergedBuffer -> {
                try (InputStream inputStream = mergedBuffer.asInputStream();
                     BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream))) {
                    return bulkImportService.importData(bufferedReader);
                } catch (IOException e) {
                    throw new RuntimeException("Failed to read file content", e);
                } finally {
                    DataBufferUtils.release(mergedBuffer);
                }
            })
            .onErrorResume(ex -> {
                log.error("[importData] An error occured during execution:", ex);
                return Mono.just(new ArrayList<>());
            })
            .doOnTerminate(() -> log.debug("[importData] END "));
}

关键注意事项 #

• 无论哪种方案，都必须手动释放DataBuffer，否则会导致内存泄漏；
• 避免在WebFlux场景中滥用block()，如果业务允许，优先选择非阻塞的异步处理方案。

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