如何从 Flux (Spring Reactor) 中的项目生成项目

Question

如何使用反应堆通量对以下场景进行建模？

  +-------+        +-------+        +-------+
  |       |        |       | -----> |       |
  |   A   | ---->  |   B   | -----> |   C   |
  |       |        |       | -----> |       |
  +-------+        +-------+        +-------+

A 生成由B 接收的项目，该项目将从该项目发出[0..N] 由C 接收的项目。

如何编写 Flux 以便 B 可以根据 A 发出的项目发出新项目？

A a = ...
B b = ...
C c = ...

Flux.push((emitter) -> a.run(emitter))
 .howToMapOneEelementToMany((emitter, item) -> b.handle(emitter, item); // ???
 .doOnNext((item) -> c.handle(item))
 .subscribe();

---

编辑：示例

• A 生成 Excel 文件的路径
• B 读取 Excel 文件并每行生成一个项目（可以是很多行或没有）
• C 将每一行存储在数据库中

Answer 1

如果您有很多文件，那么已经是 Flux。你可以用这个来做一个 flatMap，因为 Flux 的每个项目都将生成另一个行通量。 如果您想保持数据的顺序，例如 concatMap 和 flatMapSequential，还有其他方法。

 public Flux<String> someMethod() {

    return Flux.fromIterable(List.of("excelFileData", "excelFileData2"))
      .flatMap(this::getRows);

  }

  private Flux<String> getRows(String file) {
    return Flux.fromIterable(file.lines().collect(toList()));
  }

看reactor文档，挺好的。

https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Flux.html

但是，如果您想将 Mono 转换为 Flux，您可以执行以下操作：

public Flux<String> someMethod() {

  return Mono.just("excelFileData")
    .flatMapMany(this::getRows);

}

private Flux<String> getRows(String file) {
  return Flux.fromIterable(file.lines().collect(toList()));
}

Answer 2

@sfiss 的推荐 flatMap 似乎是一个不错的候选人

ExcelFilePathReaderSink pathReader = ...
ExcelRowHandlerSink rowHandler = ...
OutputWriter outWriter = ...

Flux.push(pathReader)     // A
  .flatMap(rowHandler)    // B
  .doOnNext(outWriter)    // C
  .subscribe();