我正在尝试使用Reactor的Flux并行处理多个文件。主要工作负载发生在对
每当我尝试订阅结果Flux时,主线程在我接收到任何值之前退出。
我期望在列表中找到处理流程的输出,但程序立即退出。首先我想知道我是否正确使用了Flux类,其次我应该如何等待subscribe调用完成?
flatMap
的调用中,然后对Flux进行转换和过滤。每当我尝试订阅结果Flux时,主线程在我接收到任何值之前退出。
Flux.fromStream(Files.list(Paths.get("directory"))
.flatMap(path -> {
return Flux.create(sink -> {
try (
RandomAccessFile file = new RandomAccessFile(new File(path), "r");
FileChannel fileChannel = file.getChannel()
) {
// Process file into tokens
sink.next(new Token(".."));
} catch (IOException e) {
sink.error(e);
} finally {
sink.complete();
}
}).subscribeOn(Schedulers.boundedElastic());
})
.map(token -> /* Transform tokens */)
.filter(token -> /* Filter tokens*/)
.subscribe(token -> /* Store tokens in list */)
我期望在列表中找到处理流程的输出,但程序立即退出。首先我想知道我是否正确使用了Flux类,其次我应该如何等待subscribe调用完成?