java - 如何在不丢失数据的情况下与其他发布者建立响应式管道？

假设您有这样的代码：

public List<Group> addUserToGroups(String username, String label) {
  Mono<User> userMono = webClient.getUser(username);
  User user = userMono.block();

  Flux<Group> groupsFlux = webClient.getGroups(label);
  List<Group> groups = groupsFlux.collectList().block();

  groups.forEach(group ->
      webClient.addUserToGroup(user.getId(), group.getId()).block()
  );

  return groups;
}

但是现在您想将此代码重构为非阻塞的反应式管道，并且可以使用main方法返回Flux<Group>。

因此，也许您将开始执行以下操作：

public Flux<Group> addUserToGroups(String username, String label) {
  return webClient.getUser(username)
      .flatMapMany(user -> webClient.getGroups(label))
      ...
}

但是现在我们有一个问题，结果Flux中的值是Group，我们需要在下一步中丢失的User信息。

因此，希望的管道数据流可以这样表示：

start
|
U
| /
|/
G1,U
| \
|  UG1----|
|         |
G2,U      G1
| \       |
|  UG2----|
|         |
          G2
          |
        result: G1, G2

UGn is the result of calling webClient.addUserToGroup

实施此方法的正确方法是什么？

最佳答案

通常，只要您觉得需要映射到另一个值，但又要保留已经拥有的值，就有三种基本选择：


使用嵌套的flatMap()调用；
创建一个同时包含两种类型并使用合成的新对象（或使用Tuple）；
更改对象结构或基础服务，以便“映射”对象包含您需要的所有字段。


嵌套flatMap()调用的优点是快速且容易，但缺点是在进行大量嵌套时，代码可能变得难以读取。合成解决了这个问题，但是显然您需要创建新的类型（或使用Tuple类型，这些类型显然没有那么多描述性）。

如果可行的话，更改对象结构是两全其美的做法，因此值得考虑-但根据我的经验，在大多数情况下，这样做既不可行也不明智。

恕我直言，在这里使用嵌套调用会很好，因为它只是单个嵌套级别-类似于：

return webClient.getUser(username)
        .flatMapMany(
                user -> webClient.getGroups(label)
                        .flatMap(group -> webClient.addUserToGroup(user.getId(), group.getId()).thenReturn(group))
        )
        .collectList();

不一定与问题有关，但是我不特别喜欢此方法的第二个方面-它实际上是在做两件事-为用户检索组，然后触发第二次呼叫以将该用户添加到组（功能似乎有些奇怪，但是我不知道用例。）如果可以的话，我建议将这两个功能分成两个单独的方法。