响应式编程中 Reactor Mono/Flux 实现轮询请求的策略与实践

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);
    }
}

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")