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

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

但是现在您想将此代码重构为非阻塞的反应式管道，并且是返回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

通常，只要您感觉需要映射到另一个值，但又要保留已经拥有的值，就可以使用三个基本选项：