如何在反应器 webflux 中丰富对象？答案

【问题标题】：How to do enrichment of an object in reactor webflux?如何在反应器 webflux 中丰富对象？
【发布时间】：2020-08-04 17:09:08
【问题描述】：

我有一个界面

    public interface Enricher {
       Mono<MyObject> enrich(MyObject obj);
    }

我有这个接口的实现，它返回更新后的 MyObject。

我有另一个名为 CompositeEnricher 的实现，它在构造函数中接受 Enricher 列表并执行丰富。

在非反应性的世界里，我会写这样的东西

/*
  Please note that following code is applicable only if Enricher interface returns MyObject instead of Mono<MyObject>
*/

public class CompositeEnricher implements Enricher {

  private final List<Enricher> enrichers;

  public CompositeEnricher(List<Enricher> enrichers) {
    this.enrichers = enrichers;
  }

  @Override
  public MyObject enrich(MyObject myObject) {
    MyObject updated = myObject;
    for(Enricher enricher : enrichers) {
      updated = enricher.enrich(updated);
    }
    return updated;
  }
}

如何将其更改为反应式，以便 Enricher 的每个实现都返回 Mono，包括实现 Enricher 接口的 CompositeEnricher？

【问题讨论】：

标签： java spring spring-boot spring-webflux project-reactor

【解决方案1】：

实际上根本不需要单独的 CompositeEnricher 类 - 给定 List<Enricher> enrichers，您可以使用标准 Java 流简单地减少浓缩器：

Enricher composite = enrichers.stream()
        .reduce((e1, e2) -> myObj -> e1.enrich(myObj).flatMap(myObj2 -> e2.enrich(myObj2)))
        .get();

当然，如果需要，您可以将其充实到一个类中，但您同样可以将其作为辅助方法保留在某个地方。

【讨论】：

实际上我想使用响应式 of 这样做，因为我的丰富方法返回 Mono 而不仅仅是 MyObject。
@AvikKesari 不确定你的意思 - 这里的方法仍然会给你一个 enrich() 方法，它返回一个 Mono<MyObject>?
请查看我使用 Mono.expand 发布的解决方案。我想知道我是否可以改进此解决方案并将浓缩器也放入反应链中
@AvikKesari 据我所知，您的目标是在没有优势的情况下使用反应链 - 除非浓缩器是通过反应流以某种方式获得的，而不是作为列表？
实际上，我的一个浓缩器对其他服务进行了阻塞 rmi 调用，我试图将其包装在反应链中，这样我就不会在 Blockhound 中遇到异常。为此，我已将界面更改为返回 Mono 而不是阻塞。

【解决方案2】：

我使用 Mono.expand 方法解决了这个问题。但我不得不使用enrichers.iterator()。想知道我是否可以以某种方式将浓缩器也放入反应链而不是使用迭代器。

@Override
    public Mono<User> enrich(User user) {
        Iterator<Enricher> iterator = enrichers.iterator();
        Function<User, Publisher<User>> enrichmentFunction = u -> iterator.hasNext() ? iterator.next().enrich(u) : Mono.empty();
        return Mono.just(user)
                .expand(enrichmentFunction)
                .last();
    }

【讨论】：