Spring Webflux在返回响应之前等待所有Flux

Question

我正在使用Spring Webflow R2DBC将一些数据插入数据库。

要求提供数据-

控制器

@RestController
public class UtilController {
  private final FooService service;

  @Autowired
  public UtilController(FooService service) { this.service = service; }

  @GetMapping(path = "/load-data")
  public Mono<List<Foo>> loadFoos() { return service.addFoos(); }
}

服务

public Mono<List<Foo>> addFoos() {
  return webClient.getSomeStuff()
      .map(results -> results.stream()
          .map(/* ...transform to domain model */)
          .toList())
      .map(foos -> {
        // This is the problem site.
        List<Foo> buf = new ArrayList<>();
        foos.forEach(f -> {
          fooDao.createFoo(l)
              .doOnNext(buf::add)
              .next()
              .subscribe();
        });

        return buf;
      });
}

道

public Flux<Foo> createFoo(final Foo foo) {
  var query = "<INSERT_STUFF_STATEMENT>";

  // "publisher" is private final Publisher<? extends Connection> publisher;
  var connection = Mono.from(publisher);
  return connection.flatMapMany(c -> c.createStatement(query)
      .bind("STUFF", foo.getStuff())
      .execute())
      .map(result -> Foo.builder()
          .stuff(whatever)
          .build());
}

主要问题是我不知道如何让它等待所有结果返回并添加到最终列表中




Answer 1

您是否必须使用webClient（实际上是一个没有数据库访问的外部API）？否则，我建议使用R2DBCReposition并返回通量，并从中映射数据？如果您真的想调用外部数据库，迭代列表通常是阻塞的（应该会引发错误）。此外，Lambda中的变量应该是final或实际上是final，所以我不确定发生了什么（您所称的数据库是反应型的），但仍然会尝试返回Flux而不是Mono