如何用Reactor Mono实现轮询请求？现有代码安全性及优化建议

安全性分析 #

线程安全 #

这段代码是线程安全的：

• WebClient本身为线程安全设计，可在多线程环境下共享使用；
• Reactor提供的操作符（filter、switchIfEmpty、retryWhen）均为无状态实现，不存在共享可变状态的并发修改风险；
• 每次订阅pollUntilReady时，都会重新触发webClient.get()请求，各订阅之间相互独立。

内存泄漏风险 #

正常使用场景下无内存泄漏问题：

• Retry.fixedDelay是Reactor官方实现的重试机制，会自动管理订阅生命周期，每次重试失败后都会清理相关资源；
• 只要确保订阅在达到最大重试次数或目标状态就绪后正常终止，就不会残留未释放的资源。

优化方案 #

原代码逻辑可行，但可以从语法、可读性和健壮性方面优化：

1. 修复语法错误 #

原代码中switchIfEmpty的括号未正确闭合，修正后代码如下：

Mono<Status> checkStatus = webClient.get(...)
                                   .bodyToMono(String.class)
                                   .map(Status::from);

Mono<Status> pollUntilReady = checkStatus.filter(Status::isReady)
                                        .switchIfEmpty(Mono.error(new SystemStateNotReadyException()))
                                        .retryWhen(Retry.fixedDelay(MAX_ATTEMPT, BACK_OFF)
                                                        .filter(err -> err instanceof SystemStateNotReadyException));

2. 简化状态判断逻辑 #

可以用flatMap替代filter+switchIfEmpty，直接在判断状态后返回对应信号，代码更直观：

Mono<Status> pollUntilReady = webClient.get(...)
                                      .bodyToMono(String.class)
                                      .map(Status::from)
                                      .flatMap(status -> status.isReady() 
                                          ? Mono.just(status) 
                                          : Mono.error(new SystemStateNotReadyException()))
                                      .retryWhen(Retry.fixedDelay(MAX_ATTEMPT, BACK_OFF)
                                                      .filter(err -> err instanceof SystemStateNotReadyException));

3. 增强重试可见性 #

添加重试日志，方便排查轮询过程中的问题：

Mono<Status> pollUntilReady = webClient.get(...)
                                      .bodyToMono(String.class)
                                      .map(Status::from)
                                      .flatMap(status -> {
                                          if (status.isReady()) {
                                              return Mono.just(status);
                                          }
                                          log.info("System not ready, initiating retry...");
                                          return Mono.error(new SystemStateNotReadyException());
                                      })
                                      .retryWhen(Retry.fixedDelay(MAX_ATTEMPT, BACK_OFF)
                                                      .filter(err -> err instanceof SystemStateNotReadyException)
                                                      .doBeforeRetry(signal -> log.debug("Retry attempt {} of {}", signal.totalRetries() + 1, MAX_ATTEMPT)));

4. 扩展异常重试范围 #

原代码仅重试SystemStateNotReadyException，如果需要对网络异常（如超时、连接失败）也进行重试，可以扩展过滤条件：

.retryWhen(Retry.fixedDelay(MAX_ATTEMPT, BACK_OFF)
                .filter(err -> err instanceof SystemStateNotReadyException 
                            || err instanceof WebClientResponseException 
                            || err instanceof IOException));

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