代码之家 › 专栏 › 技术社区 › Trace

始终调用Mono switchIfEmpty()

reactive spring-webflux rx-java spring-boot java

Trace · 技术社区 · 7 年前

我有两种方法。
主要方法:

@PostMapping("/login")
public Mono<ResponseEntity<ApiResponseLogin>> loginUser(@RequestBody final LoginUser loginUser) {
    return socialService.verifyAccount(loginUser)
            .flatMap(socialAccountIsValid -> {
                if (socialAccountIsValid) {
                    return this.userService.getUserByEmail(loginUser.getEmail())
                            .switchIfEmpty(insertUser(loginUser))
                            .flatMap(foundUser -> updateUser(loginUser, foundUser))
                            .map(savedUser -> {
                                String jwts = jwt.createJwts(savedUser.get_id(), savedUser.getFirstName(), "user");
                                return new ResponseEntity<>(HttpStatus.OK);
                            });
                } else {
                    return Mono.just(new ResponseEntity<>(HttpStatus.UNAUTHORIZED));
                }
            });

}

这个被调用的方法(服务调用外部api):

public Mono<User> getUserByEmail(String email) {
    UriComponentsBuilder builder = UriComponentsBuilder
            .fromHttpUrl(USER_API_BASE_URI)
            .queryParam("email", email);
    return this.webClient.get()
            .uri(builder.toUriString())
            .exchange()
            .flatMap(resp -> {
                if (Integer.valueOf(404).equals(resp.statusCode().value())) {
                    return Mono.empty();
                } else {
                    return resp.bodyToMono(User.class);
                }
            });
}

在上面的例子中, switchIfEmpty() 始终从主方法调用,即使结果为 Mono.empty() 他回来了。

我找不到解决这个简单问题的办法。
以下内容也不起作用:

Mono.just(null)

因为该方法将抛出nullpointerexception。

我也不能使用flatMap方法来检查它 foundUser 是空的。
Mono.empty() ,因此我也不能在这里添加条件。

感谢您的帮助。

   @PostMapping("/login")
    public Mono<ResponseEntity<ApiResponseLogin>> loginUser(@RequestBody final LoginUser loginUser) {
        userExists = false;
        return socialService.verifyAccount(loginUser)
                .flatMap(socialAccountIsValid -> {
                    if (socialAccountIsValid) {
                        return this.userService.getUserByEmail(loginUser.getEmail())
                                .flatMap(foundUser -> {
                                    return updateUser(loginUser, foundUser);
                                })
                                .switchIfEmpty(Mono.defer(() -> insertUser(loginUser)))
                                .map(savedUser -> {
                                    String jwts = jwt.createJwts(savedUser.get_id(), savedUser.getFirstName(), "user");
                                    return new ResponseEntity<>(HttpStatus.OK);
                                });
                    } else {
                        return Mono.just(new ResponseEntity<>(HttpStatus.UNAUTHORIZED));
                    }
                });

    }

1 回复 | 直到 7 年前

Alex 7 年前

这是因为switchIfEmpty“按值”接受Mono。这意味着,即使在您订阅mono之前,此替代mono的评估也已经触发。

想象这样一种方法:

Mono<String> asyncAlternative() {
    return Mono.fromFuture(CompletableFuture.supplyAsync(() -> {
        System.out.println("Hi there");
        return "Alternative";
    }));
}

如果您这样定义代码:

Mono<String> result = Mono.just("Some payload").switchIfEmpty(asyncAlternative());

无论在溪流施工期间发生什么情况,它都会触发备选方案。为了解决这个问题,您可以使用 Mono.defer

Mono<String> result = Mono.just("Some payload")
        .switchIfEmpty(Mono.defer(() -> asyncAlternative()));

这样,它只会在请求替代方案时打印“Hi there”

UPD:

单声道<字符串>result=Mono.just(“一些有效负载”).switchIfEmpty(asyncAlternative());

我们可以将其改写为:

Mono<String> firstMono = Mono.just("Some payload");
Mono<String> alternativeMono = asyncAlternative();
Mono<String> result = firstMono.switchIfEmpty(alternativeMono);

这两个代码段在语义上是等价的。我们可以继续展开它们以查看问题所在:

Mono<String> firstMono = Mono.just("Some payload");
CompletableFuture<String> alternativePromise = CompletableFuture.supplyAsync(() -> {
        System.out.println("Hi there");
        return "Alternative";
    }); // future computation already tiggered
Mono<String> alternativeMono = Mono.fromFuture(alternativePromise);
Mono<String> result = firstMono.switchIfEmpty(alternativeMono);

正如您所看到的,未来的计算在我们开始编写 Mono 类型。为了避免不必要的计算,我们可以将我们的未来包装成延迟评估:

单声道<字符串>结果=Mono.just(“一些有效负载”)
.switchIfEmpty(Mono.defer(()->asyncAlternative());

它将展开成

Mono<String> firstMono = Mono.just("Some payload");
Mono<String> alternativeMono = Mono.defer(() -> Mono.fromFuture(CompletableFuture.supplyAsync(() -> {
        System.out.println("Hi there");
        return "Alternative";
    }))); // future computation defered
Mono<String> result = firstMono.switchIfEmpty(alternativeMono);

在第二个例子中,未来被困在一个懒惰的供应商中,只有在需要时才被安排执行。

Philippe Simo 5 年前

对于那些尽管投票结果很好,但仍不明白为何会有这种行为的人:

反应器源(Mono.xxx和Flux.xxx)为:

:仅当订户订阅源内容时,才会评估/触发源内容;
或 热切评价 :即使在订阅者订阅之前,也会立即评估源的内容。

表达方式如 Mono.just(xxx) Flux.just(xxx) , Flux.fromIterable(x,y,z) 你很渴望。

利用 defer()

这样做:

 someMethodReturningAMono()
  .switchIfEmpty(buildError());

buildError() 依靠渴望的来源来创建替代单声道永远在订阅之前进行评估:

Mono<String> buildError(){
       return Mono.just("An error occured!"); //<-- evaluated as soon as read
}

要防止出现这种情况,请执行以下操作:

 someMethodReturningAMono()
  .switchIfEmpty(Mono.defer(() -> buildError()));

读这个 answer 更多。