Zip reactive flow自身

Question

我使用的是Java Reactor Core，并且我有一个反应式的对象Flux。对于Flux的每个对象，我需要执行一个外部查询，为每个输入返回一个不同的对象。然后，需要将新生成的Flux与原始Flux一起压缩-因此，必须同步并以相同的顺序生成2个Flux的项目。

我只是重复使用同一个流两次，如下所示：

 Flux<MyObj> aTest = Flux.fromIterable(aListOfObj);

Flux<String> myObjLists = aTest.map(o -> MyRepository.findById(o.name)).map(o -> {
  if (!o.isPresent()) {
    System.out.println("Fallback to empty-object");
    return "";
  }
  List<String> l = o.get();
  if (l.size() > 1) {
    System.out.println("that's bad");
  }
  return l.get(0);
});

Flux.zip(aTest, myObjLists, (a, b) -> doSomethingWith(a,b))

这是正确的做法吗？如果myObjLists发出错误，如何防止zip阶段跳过失败的迭代？

Answer 1

我最终选择了使用Tuples和Optionals (以防止会破坏流量的空项目)，这样我就不需要重用初始的Flux

Flux<Tuple<MyObj, Optional<String>>> myObjLists = Flux.fromIterable(aListOfObj)
  .map(o -> Tuples.of(o, Optional.ofNullable(MyRepository.findById(o.name))
  .flatMap(t -> {
    if (!t.getT2().isPresent()) {
      System.out.println("Discarding this item");
      return Flux.empty();
    }
    List<String> l = t.getT2().get();
    if (l.size() > 1) {
      System.out.println("that's bad");
    }
    return Tuples.of(t.getT1(), l.get(0));
  })
  .map(t -> doSomethingWith(t.getT1(),t.getT2()))

请注意，可以用.map().filter()替换flatMap，删除缺少可选项目的元组