使用CSVReader响应式处理ZIP中CSV文件的异常问题排查

核心原因 #

问题出在ZipInputStream未正确初始化：ZipInputStream需要先调用getNextEntry()方法定位到压缩包内的具体文件条目，否则流不会开始读取实际的文件内容。CSVReader基于未初始化的ZipInputStream创建时，无法读取到任何数据，导致collectList()要么得到空列表，要么因等待元素而卡住。

你觉得“直接订阅Flux正常”，大概率是因为订阅逻辑没有等待所有元素处理完成（比如未添加block()或doOnComplete验证），控制台无错误输出让你误以为流程正常，但实际上并没有读取到CSV数据。

修复方案 #

在创建CSVReader之前，先调用ZipInputStream#getNextEntry()定位到压缩包内的CSV文件：

@SneakyThrows
Mono<Void> csvFlux() {
    var failure = new AtomicInteger();
    var filtered = new ArrayList<String>();
    var is = new ByteArrayInputStream(createZipContent(CSV_FILE_NAME, TEST_CSV.getBytes(StandardCharsets.UTF_8)));
    var zipInputStream = new ZipInputStream(is);
    
    // 关键步骤：定位到压缩包内的CSV文件条目
    ZipEntry csvEntry = zipInputStream.getNextEntry();
    if (csvEntry == null || !csvEntry.getName().equals(CSV_FILE_NAME)) {
        return Mono.error(new IllegalArgumentException("压缩包内未找到目标CSV文件"));
    }
    
    var reader = new CSVReader(new InputStreamReader(zipInputStream));
    return Flux.fromIterable(reader)
            .flatMap(this::ean)
            .filter(ean -> filter(ean, filtered))
            .onErrorContinue((ex, integer) -> failure.getAndIncrement())
            .collectList()
            .flatMap(list -> print(list.size(), failure, filtered));
}

额外说明 #

1. CSVReader的迭代器特性：CSVReader实现的Iterable是基于流的一次性迭代器，每次调用next()都会从输入流读取一行数据，流的状态会直接影响迭代结果。
2. ZipInputStream的特性：ZipInputStream是顺序流，必须通过getNextEntry()切换到具体的压缩文件，否则无法读取到文件内容；且流不支持回退，迭代器只能遍历一次。

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