Webflux流式处理DataBuffer生成Zip文件时出现损坏问题

核心问题点 #

你的代码存在三个关键问题，导致生成的Zip文件损坏且条目重复：

1. 复用单一DefaultDataBuffer：全程只创建了一个缓冲区实例，每次处理完文件后都返回它，导致Flux多次发射同一个缓冲区的内容，最终浏览器接收到重复的Zip条目数据。
2. 阻塞IO与WebFlux非阻塞模型冲突：ZipOutputStream是阻塞式IO组件，直接在Flux的map操作中调用其方法会阻塞WebFlux的事件循环线程，破坏非阻塞设计，同时可能导致数据写入不及时。
3. 未正确处理Zip收尾数据：在doOnComplete中关闭ZipOutputStream时，缓冲区的收尾数据（如Zip的目录结构）无法被正确发射到浏览器，导致Zip文件损坏。

修复后的代码 #

import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import reactor.util.function.Tuple2;

import java.io.IOException;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.zip.ZipEntry;
import java.util.zip.ZipOutputStream;

@RestController
public class ZipController {

    @GetMapping(value = "/zip", produces = MediaType.APPLICATION_OCTET_STREAM_VALUE)
    public Flux<DataBuffer> zip() {
        var files = Arrays.asList("File1", "File2", "File3", "File4", "File5");
        DataBufferFactory bufferFactory = new DefaultDataBufferFactory();

        // 使用管道流桥接阻塞ZipOutputStream与反应式Flux
        return Flux.using(
            // 初始化管道资源
            () -> {
                PipedOutputStream pos = new PipedOutputStream();
                PipedInputStream pis = new PipedInputStream(pos);
                return new Tuple2<>(pis, pos);
            },
            // 处理Zip生成与流式返回
            tuple -> {
                PipedInputStream pis = tuple.getT1();
                PipedOutputStream pos = tuple.getT2();

                // 在阻塞线程池执行Zip写入，避免占用事件循环线程
                return Mono.fromRunnable(() -> {
                    try (ZipOutputStream zipOut = new ZipOutputStream(pos)) {
                        for (String file : files) {
                            zipOut.putNextEntry(new ZipEntry(file + ".txt"));
                            zipOut.write(file.getBytes(StandardCharsets.UTF_8));
                            zipOut.closeEntry();
                        }
                        zipOut.flush(); // 确保所有数据写入管道
                    } catch (IOException e) {
                        throw new RuntimeException("生成Zip文件失败", e);
                    }
                }).subscribeOn(Schedulers.boundedElastic())
                  .thenMany(DataBufferUtils.read(pis, bufferFactory, 8192));
            },
            // 清理管道资源
            tuple -> {
                try {
                    tuple.getT1().close();
                    tuple.getT2().close();
                } catch (IOException ignored) {}
            }
        );
    }
}

修复说明 #

1. 管道流桥接阻塞与非阻塞：用PipedInputStream和PipedOutputStream将阻塞的ZipOutputStream与WebFlux的反应式数据流打通，实现数据的流式传递。
2. 线程隔离：将Zip写入操作放到Schedulers.boundedElastic()线程池，避免阻塞WebFlux的事件循环，符合非阻塞IO的设计原则。
3. 资源生命周期管理：通过Flux.using自动管理管道流的创建与关闭，避免资源泄漏。
4. 流式数据发射：使用DataBufferUtils.read将管道输入流转换为Flux<DataBuffer>，让浏览器可以逐步接收数据，不会出现重复条目或损坏问题。
5. 明确编码：指定StandardCharsets.UTF_8写入文件内容，避免平台默认编码导致的乱码问题。

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