java - 如何在不丢失数据的情况下与不同的发布者建立 react 性管道？

Question

假设您有这样的代码:

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;
}

但是现在你想将这段代码重构为一个非阻塞的响应式(Reactive)管道，并且 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

什么是正确的实现方式？

Answer 1

一般来说，每当您觉得需要映射到另一个值，但又要保留已有的值时，您有以下三种基本选择:

• 使用嵌套的 flatMap() 调用；
• 创建一个新对象来保存这两种类型并使用组合(或使用元组)；
• 更改您的对象结构或底层服务，以便“映射”对象包含您需要的所有字段。

嵌套 flatMap() 调用的优点是它快速且简单，但缺点是如果嵌套级别过多，代码可能变得几乎不可读。组合解决了这个问题，但是您显然需要创建新类型(或使用 Tuple 类型，这显然不太具有描述性。)

更改对象结构是两全其美的方法如果这样做有意义，因此值得考虑 - 但大多数时候，根据我的经验，这是不可行或不明智的。

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

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

不一定与问题相关，但还有第二个方面我不太喜欢这种方法 - 它实际上做了两件事 - 它为用户检索组，然后然后触发关闭第二次调用以将用户添加到组中(功能似乎有点奇怪，但我不知道用例。)如果可以的话，我建议将这两个功能分成两个单独的方法。