Java Stream 与 Flux fromIterable

Question

我有一个用户名列表，想在不阻塞主线程的情况下从远程服务中获取用户详细信息。我正在使用 Spring 的响应式(Reactive)客户端 WebClient。对于响应，我得到 Mono 然后订阅它并打印结果。

private Mono<User> getUser(String username) {
    return webClient
            .get()
            .uri(uri + "/users/" + username)
            .retrieve()
            .bodyToMono(User.class)
            .doOnError(e -> 
                logger.error("Error on retrieveing a user details {}", username));
}

我通过两种方式实现了任务:

使用 Java stream

usernameList.stream()
          .map(this::getUser)
          .forEach(mono ->
                mono.subscribe(System.out::println));

使用 Flux.fromIterable :

Flux.fromIterable(usernameList)
        .map(this::getUser)
        .subscribe(mono ->
                mono.subscribe(System.out::println));

似乎主线程并没有以两种方式阻塞。 Java Stream有什么区别和 Flux.fromIterable在这种情况下？如果两者都做同样的事情，建议使用哪一个？

Answer 1

两种变体之间没有太大差异。 Flux.fromIterable 变体可能会为您提供更多关于并发/重试等的选择和控制 - 但在这种情况下并非如此，因为在此处调用 subscribe 违背了目的。

您的问题缺少有关您正在构建的应用程序类型以及这些调用是在何种上下文中进行的一些背景知识。如果您正在构建一个 Web 应用程序并且在请求处理或批处理应用程序期间调用它 - 意见可能会有所不同。

一般来说，我认为应用程序应该远离调用 subscribe，因为它会断开该管道的处理与应用程序的其余部分的连接:如果发生异常，您可能无法报告它，因为用于发送该错误消息的资源那个时候可能就没了。或者也许应用程序正在关闭，而您无法让它等待该任务的完成。

如果您正在构建一个想要开始一些工作的应用程序并且其结果对当前操作没有用(即在当前操作的生命周期内该工作是否完成并不重要)，那么 subscribe 可能成为一种选择。

在这种情况下，我会尝试将所有操作组合在一个 Mono<Void> 操作中，然后触发该工作:

Mono<Void> logUsers = Flux.fromIterable(userNameList)
    .map(name -> getUser(name))
    .doOnNext(user -> System.out.println(user)) // assuming this is non I/O work
    .then();
logUsers.subscribe(...);

如果您担心在 Web 应用程序中使用服务器线程，那么情况就完全不同了——您可能希望获得该操作的结果以向 HTTP 响应写入一些内容。通过调用 subscribe，这两个任务现在都已断开连接，并且在工作完成时 HTTP 响应可能早已消失(并且在写入响应时会出现错误)。

在这种情况下，您应该将操作与 Reactor 运算符链接起来。