Zip反应流与自身

Question

我使用的是JavaReactor核心，我有一个反应式的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))

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

Answer 1

我最终选择了使用元组和可选项（以防止空项破坏通量），这样我就不需要重复使用初始通量：

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（），删除缺少可选项的元组