本教程探讨了在 Reactor 响应式编程中实现外部系统状态轮询的两种主要策略。首先介绍基于 `Mono` 的 `retryWhen` 机制，适用于在错误发生时重试。随后深入讲解利用 `Flux.interval` 实现固定间隔轮询的方法，该方法在控制请求频率、并发性及避免异常流控制方面具有优势，并提供了详细的代码示例和选择考量。
引言：响应式轮询的需求
在构建现代分布式系统时，我们经常需要与外部服务交互，并等待其状态从“处理中”变为“就绪”。这种持续检查外部状态直到满足特定条件的行为，通常通过轮询（Polling）机制实现。在基于 Project Reactor 的响应式编程范式中，实现高效、健壮且资源友好的轮询策略是关键。本文将深入探讨两种主要的 Reactor 轮询实现方法：基于 retryWhen 的错误重试机制和基于 Flux.interval 的固定间隔轮询策略。
基于 retryWhen 的轮询实现
最初，开发者可能会倾向于使用 Reactor 的 retryWhen 操作符来处理轮询场景。这种方法的核心思想是：当外部系统状态不满足条件时，通过抛出特定异常来触发重试机制，从而实现周期性的状态检查。
实现原理
该方法通常结合 WebClient 发起请求，并通过 filter 操作符检查返回的状态。如果状态不符合预期（例如，系统尚未就绪），则使用 switchIfEmpty 结合 Mono.error() 抛出一个自定义异常。随后，retryWhen 操作符捕获此异常，并根据配置的重试策略（如固定延迟）重新订阅上游流，再次发起请求。
以下是一个典型的实现示例：
import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;
import org.springframework.web.reactive.function.client.WebClient;
import java.time.Duration;
// 假设 Status 是一个包含 isReady() 方法的枚举或类
// 假设 SystemStateNotReadyException 是一个自定义异常
public class PollingWithRetryWhen {
private final WebClient webClient;
private final int MAX_ATTEMPT = 5; // 最大重试次数
private final Duration BACK_OFF = Duration.ofSeconds(1); // 重试间隔
public PollingWithRetryWhen(WebClient webClient) {
this.webClient = webClient;
}
/**
* 模拟从外部系统获取状态。
* 实际应用中会通过 webClient 调用外部API。
*/
private Mono<Status> checkStatus() {
return webClient.get()
.uri("/api/status")
.retrieve()
.bodyToMono(String.class)
.map(response -> Status.from(response)); // 假设 Status.from(String) 能够解析响应
}
/**
* 轮询外部系统直到其状态变为就绪。
* 如果未就绪，则抛出异常并重试。
*/
public Mono<Status> pollUntilReady() {
return checkStatus()
.filter(Status::isReady) // 过滤出已就绪的状态
.switchIfEmpty(
Mono.error(new SystemStateNotReadyException("System is not ready yet.")) // 如果未就绪，则抛出异常
)
.retryWhen(
Retry.fixedDelay(MAX_ATTEMPT, BACK_OFF) // 固定延迟重试
.filter(err -> err instanceof SystemStateNotReadyException) // 只对特定异常进行重试
);
}
}
// 示例 Status 类，代表外部系统状态
class Status {
private final boolean ready;
public Status(boolean ready) {
this.ready = ready;
}
public boolean isReady() {
return ready;
}
public static Status from(String response) {
// 模拟解析逻辑，例如根据响应字符串判断是否就绪
return new Status(response.contains("READY"));
}
}
// 示例自定义异常，用于在系统未就绪时触发重试
class SystemStateNotReadyException extends RuntimeException {
public SystemStateNotReadyException(String message) {
super(message);
}
}优点与考量
- 简洁性： 对于简单的重试逻辑，retryWhen 提供了一种声明式且易于理解的方式。
- 错误驱动： 这种方法天然地将轮询与错误处理相结合，只有在状态不满足条件（被视为一种“错误”）时才触发重试。
- 线程安全与内存： Reactor 框架本身在设计时就考虑了线程安全和资源管理。上述代码片段在典型使用场景下通常是线程安全的，并且不会导致内存泄漏，因为 Reactor 的操作符会妥善管理订阅和资源。
然而，retryWhen 的核心是“重试”，这意味着它通常在操作失败（抛出异常）后才触发。如果每次轮询请求本身是成功的（即返回了状态，只是状态值不符合预期），但我们仍想以固定间隔进行检查，那么通过抛出异常来控制流程可能不是最优雅或最高效的方式。
基于 Flux.interval 的高级轮询策略
当需要以精确的固定间隔进行轮询，并且希望更精细地控制轮询过程（例如，独立于每次请求的响应时间），或者避免使用异常来控制正常流程时，Flux.interval 提供了一个更强大的替代方案。
实现原理
Flux.interval 会周期性地发出递增的 Long 值，每个值代表一个时间间隔。我们可以将这些周期性信号映射（concatMap 或 flatMap）到我们的状态检查请求中。通过这种方式，即使某个状态检查请求耗时较长，下一个请求也会在预设的固定间隔后触发，而不是等待上一个请求完全完成后再计算延迟。
优点
- 固定间隔： 确保请求以固定的时间间隔发送，独立于单个请求的响应时间。
- 计数器： Flux.interval 发出的 Long 值可以作为轮询尝试的计数器。
- 并发控制： 可以选择 concatMap（顺序执行，等待前一个完成）或 flatMap（并发执行，不等待前一个完成）来控制状态检查请求的并发行为。
- 避免异常流控制： 不需要通过抛出和捕获异常来控制轮询流程，这在某些情况下可能带来性能优势。
示例代码
以下是一个使用 Flux.interval 实现固定间隔轮询的示例：
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.web.reactive.function.client.WebClient;
import java.time.Duration;
import java.util.concurrent.TimeoutException; // 用于可能的超时异常
public class PollingWithFluxInterval {
private final WebClient webClient;
private final int MAX_ATTEMPTS = 10; // 最大轮询尝试次数
private final Duration INTERVAL = Duration.ofMillis(100); // 每100毫秒发送一次请求
public PollingWithFluxInterval(WebClient webClient) {
this.webClient = webClient;
}
/**
* 模拟从外部系统获取状态，返回一个包含尝试次数和状态就绪情况的报告。
* 模拟了每次请求的网络延迟。
*/
private Mono<Report> fetchStatus(long count) {
// 模拟网络请求和处理时间，例如50ms
return webClient.get()
.uri("/api/status") 
