如何使用InputStreamand OutputStream以反应式方式读取和写入文件

Question

我试图在操作中读取Excel文件，或向其中添加新数据并将其写回。我还试图用通量和Mono来完成一个完整的反应过程。其思想是通过webservice返回结果文件或bytearray。

我的问题是如何以非阻塞方式获取InputStream和OutputStream？

我正在使用Apache Poi库读取并生成Excel文件。

我目前有一个基于混合Mono.fromCallable（）和阻塞代码获取输入流的解决方案。

例如，webservice部分如下所示。

@GetMapping(value = API_BASE_PATH + "/download", produces = "application/vnd.ms-excel")
public Mono<ByteArrayResource> download() {
    Flux<TimeKeepingEntry> createExcel = excelExport.createDocument(false);

    return createExcel.then(Mono.fromCallable(() -> {
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        excelExport.getWb().write(outputStream);
        return new ByteArrayResource(outputStream.toByteArray());
    }).subscribeOn(Schedulers.elastic()));
}

以及文件的处理：

public Flux<TimeKeepingEntry> createDocument(boolean all) {
    Flux<TimeKeepingEntry> entries = null;
    try {
        InputStream inputStream = new ClassPathResource("Timesheet Template.xlsx").getInputStream();
        wb = WorkbookFactory.create(inputStream);
        Sheet sheet = wb.getSheetAt(0);

        log.info("Created document");

        if (all) {
            //all entries
        } else {
            entries = service.findByMonth(currentMonthName).log("Excel Export - retrievedMonths").sort(Comparator.comparing(TimeKeepingEntry::getDateOfMonth)).doOnNext(timeKeepingEntry-> {
                this.populateEntry(sheet, timeKeepingEntry);
            });
        }
    } catch (IOException e) {
        log.error("Error Importing File", e);
    }
    return entries;
}

这可以很好地工作，但与Flux和Mono不太一致。这里有一些指导会很好。我希望整个序列都是非阻塞的。

Answer 1

不幸的是，WorkbookFactory。create（）操作正在阻塞，因此必须使用命令式代码执行该操作。然而，获取每个timeKeepingEntry可以反应性地完成。您的代码如下所示：

public Flux<TimeKeepingEntry> createDocument() {
    return Flux.generate(
        this::getWorkbookSheet,
        (sheet, sink) -> {
            sink.next(getNextTimeKeepingEntryFrom(sheet));
        },
        this::closeWorkbook);
}

这会将工作簿保存在内存中，但在请求通量元素时，会根据需要获取每个条目。