关于如何使用 Project-Reactor 库将遗留 Java 代码转换为 Reactive 的问题

Question

我打算用 project-reactor 库编写一个 process() 方法

process() 方法以字节数组为参数，经过以下步骤。假设每个步骤的所有其他方法都已编写好并可以使用。

1. 将字节数组反序列化为 Message 对象
2. 从 Message 对象中提取 userID、userStatus 和 userAddress
3. 根据 userId 从数据库中检索记录
4. 如果从 Message 对象中提取的这两个值不是 Null，则使用 userStatus 和 userAddress 更新记录。
5. 将记录保存在数据库中

@Component
public class UserService {

    @Autowired
    private Repository repo;

    //this is the method I want to fix
    public Mono<User> process(byte[] byteArray) {
        Message msg = deserialize(byteArray).block();  --> Question #1
        String userId = extractUserId(msg);
        String userStatus = extractUserStatus(msg);
        String userAddress = extractUserAddress(msg);
        return repo.find(userId)
                .switchIfEmpty(Mono.defer(() -> {
                    log.error("Error Message");
                    return Mono.empty();
                }))
                .map(user -> {
                    if(userStatus != null) {
                        user.setStatus(userStatus);
                    }

                    if(userAddress != null) {
                        user.setAddress(userAddress);
                    }
                })
                .flatmap(repo::save);
    }

    private Mono<Message> deserialize(byte[] byteArray) {
        //assume this method is written and is ready to be invoked;
    }

    private String extractUserId(Message msg) {
        //assume this method is written and is ready to be invoked;
    }

    private String extractUserStatus(Message msg) {
        //assume this method is written and is ready to be invoked;
    }

    private String extractUserAddress(Message msg) {
        //assume this method is written and is ready to be invoked;
    }

}

public class Repository {
    public Mono<User> find (String id) {
        //assume this method is written and is ready to be invoked;
    }

    public Mono<User> save(User user) {
        //assume this method is written and is ready to be invoked;
    }
}

问题 1：我不应该阻止它，但我必须从中提取 ID 以从数据库中检索记录，并在以后使用它来提取 userStatus 和地址。

问题 2：extractUserId、extractUserStatus 和 extractUserAddress 方法是否应该返回 Mono？

Answer 1

问题 1：我不应该阻止它，但是...

这并不是你应该阻止它，但你不能阻止它（除非你采用诸如为特定部分设置线程池之类的策略在你别无选择的链中，这很混乱，并且通常表明你还没有准备好移动到完全反应式堆栈。）在完全反应式链中，任何块都会杀死链死，捆绑你的宝贵的事件循环线程和严重的性能。

相反，您可以将其保留为 Mono<Message>，然后使用它做您需要做的事情 - 因此对于您当前的实现，它可能看起来像这样：

return deserialize(byteArray)
    .flatMap(
        msg -> {
          String userId = extractUserId(msg);
          String userStatus = extractUserStatus(msg);
          String userAddress = extractUserAddress(msg);

          return repo.find(userId)
              .switchIfEmpty(
                  Mono.defer(
                      () -> {
                        log.error("Error Message");
                        return Mono.empty();
                      }))
              .map(
                  user -> {
                    if (userStatus != null) {
                      user.setStatus(userStatus);
                    }

                    if (userAddress != null) {
                      user.setAddress(userAddress);
                    }
                  })
              .flatmap(repo::save);
        });

extractUserId、extractUserStatus 和 extractUserAddress 方法是否应该返回 Mono？

这取决于这些方法需要做什么。如果他们不返回单声道，他们是否必须阻止？如果是，那么他们应该返回 Mono（而不是阻塞）。如果他们不需要阻塞，那么他们没有理由返回 mono（实际上他们不应该，因为这可能导致开发人员做出错误的假设关于那些方法。）

请注意，deserialize() 方法也是如此。如果它采用字节数组，那么它可能不需要执行任何阻塞操作（反序列化阻塞的通常原因是因为它是从输入流或类似内容中读取，而不是因为实际的反序列化本身。）