Spring Boot Reactive注册接口：组合Mono返回AuthResponse失败排查

我正在编写一个注册接口，期望返回包含id和token的AuthResponse DTO。尝试在控制器中用flatMap替代map但结果一致，代码分为注册（保存并校验用户）和查询用户ID两部分，现在的问题是不知道怎么映射涉及两个Mono变量的结果，而且运行时出现“finding fail”错误，查不到刚注册的用户。

控制器代码 #

@PostMapping("/signup")
public Mono<AuthResponse> signup(@RequestBody user_info user) {
    return tokenService.signup(user)
            .doOnNext(System.out::println)
            .switchIfEmpty(Mono.error(new Error("signup not work")))
            .flatMap(token->{
                return userRespository.findByyUsername(user.getUsername())
                        .doOnNext(System.out::println)
                        .switchIfEmpty(Mono.error(new Error("finding fail")))
                        .flatMap(userInfo ->
                                Mono.just(new AuthResponse(userInfo.getId(),token)));
            });

AuthResponse（DTO） #

@Data
@AllArgsConstructor
public class AuthResponse {
    private Integer id;
    private String token;
}

错误输出 #

2023-02-05T02:45:41.539+08:00 ERROR 147274 --- [tor-tcp-epoll-2] a.w.r.e.AbstractErrorWebExceptionHandler : [c841e0d1-1]  500 Server Error for HTTP POST "/signup"

java.lang.Error: finding fail
    at com.springboot.sohinalex.java.Controller.apigatewayController.lambda$signup$1(apigatewayController.java:51) ~[classes/:na]
    Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Error has been observed at the following site(s):
    *__checkpoint ⇢ Handler com.springboot.sohinalex.java.Controller.apigatewayController#signup(user_info) [DispatcherHandler]
    *__checkpoint ⇢ org.springframework.cloud.gateway.filter.WeightCalculatorWebFilter [DefaultWebFilterChain]
    *__checkpoint ⇢ org.springframework.security.web.server.authorization.AuthorizationWebFilter [DefaultWebFilterChain]
    *__checkpoint ⇢ org.springframework.security.web.server.authorization.ExceptionTranslationWebFilter [DefaultWebFilterChain]
    *__checkpoint ⇢ org.springframework.security.web.server.authentication.logout.LogoutWebFilter [DefaultWebFilterChain]
    *__checkpoint ⇢ org.springframework.security.web.server.savedrequest.ServerRequestCacheWebFilter [DefaultWebFilterChain]
    *__checkpoint ⇢ org.springframework.security.web.server.context.SecurityContextServerWebExchangeWebFilter [DefaultWebFilterChain]
    *__checkpoint ⇢ org.springframework.security.web.server.authentication.AuthenticationWebFilter [DefaultWebFilterChain]
    *__checkpoint ⇢ org.springframework.security.web.server.authentication.AuthenticationWebFilter [DefaultWebFilterChain]
    *__checkpoint ⇢ org.springframework.security.web.server.context.ReactorContex

signup函数实现 #

public Mono<String> signup(user_info user)  {

            log.info("signup start");
            return Mono.just(user).doOnNext(System.out::println)
                    .switchIfEmpty(Mono.error(new RuntimeException()))
                         .flatMap(Monouser->{
                     Mono<Boolean> isuserexist=IsUsernameExist(Monouser.getUsername())
                             .doOnNext(System.out::println)
                             .switchIfEmpty(Mono.just(false))  //no user found => can register
                             .mapNotNull(res-> res
                             );
                     Monouser.setPassword(  //encode the password
                                     passwordEncoder.encode(user.getPassword()));
                     Mono<user_info> savedusr = respository.save(
                     Monouser).log();

                            return isuserexist.doOnNext(System.out::println) //check the username exist in my db
                                    .switchIfEmpty(Mono.error(new RuntimeException()))
                                    .flatMap(
                                    res->{
                                        log.info("start the map");
                                        if(res){
                                        log.info("error");
                                        return null;
                                    }
                                    else {
                                            savedusr.subscribe();//execute the saving user
                                            log.info("check");
                                       return  reactiveAuthenticationManager.authenticate(
                                                new UsernamePasswordAuthenticationToken(
                                                        user.getUsername(),user.getPassword()
                                                )
                                        ).map(this::generateToken);

                                    }
                                    }
                            );
}

核心问题 #

1. 异步操作脱离响应式流：signup函数中调用savedusr.subscribe()手动订阅保存用户的操作，会让该操作脱离当前响应式流的上下文，导致控制器查询用户时，保存操作可能还未完成，自然查不到数据，触发“finding fail”错误。
2. 时序与映射逻辑混乱：控制器需要同时获取用户ID和token，但当前流程是先拿token再查用户，且保存操作未与token生成串联，导致数据不一致。

修复步骤 #

1. 重构signup函数，串联完整响应式流 #

移除手动订阅，将保存用户、认证、生成token的操作串联成一个完整的响应式流，确保保存完成后再执行后续步骤，同时直接返回AuthResponse，避免二次查询。

修改后的signup函数：

public Mono<AuthResponse> signup(user_info user) {
    log.info("signup start");
    return Mono.just(user)
            .doOnNext(System.out::println)
            .switchIfEmpty(Mono.error(new RuntimeException("用户信息不能为空")))
            .flatMap(monoUser -> {
                // 检查用户名是否存在
                return IsUsernameExist(monoUser.getUsername())
                        .doOnNext(System.out::println)
                        .switchIfEmpty(Mono.just(false))
                        .flatMap(isExist -> {
                            if (isExist) {
                                log.info("用户名已存在");
                                return Mono.error(new RuntimeException("用户名已存在"));
                            }
                            // 加密密码
                            monoUser.setPassword(passwordEncoder.encode(user.getPassword()));
                            // 保存用户后直接串联认证、生成token流程
                            return respository.save(monoUser)
                                    .log()
                                    .flatMap(savedUser -> 
                                        reactiveAuthenticationManager.authenticate(
                                                new UsernamePasswordAuthenticationToken(
                                                        savedUser.getUsername(), user.getPassword()
                                                )
                                        )
                                        .map(this::generateToken)
                                        .map(token -> new AuthResponse(savedUser.getId(), token))
                                    );
                        });
            });
}

2. 简化控制器代码 #

由于signup函数已直接返回Mono<AuthResponse>，控制器无需再二次查询用户：

@PostMapping("/signup")
public Mono<AuthResponse> signup(@RequestBody user_info user) {
    return tokenService.signup(user)
            .doOnNext(System.out::println)
            .switchIfEmpty(Mono.error(new RuntimeException("注册失败")));
}

额外优化点 #

• 避免使用Error类，改用RuntimeException或自定义业务异常，更符合Spring WebFlux异常处理规范。
• IsUsernameExist可直接替换为respository.existsByUsername(username)，简化代码逻辑。
• 响应式流中禁止返回null，错误场景直接返回Mono.error()。

内容的提问来源于stack exchange，提问作者user20112858