本文详解 Reactor 中轮询逻辑无法终止的根本原因，指出 repeatWhen 与 repeat(n) 组合导致无限流的问题，并提供基于 retryWhen 的可终止、可控重试方案，包含延迟策略、失败判定与边界控制。

本文详解 reactor 中轮询逻辑无法终止的根本原因，指出 `repeatwhen` 与 `repeat(n)` 组合导致无限流的问题，并提供基于 `retrywhen` 的可终止、可控重试方案，包含延迟策略、失败判定与边界控制。

在使用 Project Reactor 实现轮询（polling）时，一个常见误区是误用 repeatWhen 与 repeat(n) 的组合——这极易导致流永不终止。如问题代码所示：

Mono.defer(() -> webClient.getResponse())
    .repeatWhen(repeat -> repeat.delayElements(Duration.ofMillis(500)))
    .repeat(4)
    .takeUntil(response -> response.equals("1"))
    .log()
    .subscribe(...);

该逻辑存在两个关键缺陷：

1. repeatWhen 优先级更高且默认无限：repeatWhen 会持续监听其内部 Publisher 的信号；若未显式终止（例如通过 take(n) 或 timeout()），它将不断触发源 Mono 的重新订阅，形成无限循环；
2. repeat(4) 作用对象错误：它重复的是 已由 repeatWhen 改造后的 Flux，而非原始 Mono，因此实际执行远超 4 次；
3. takeUntil 失效：因每次 repeatWhen 触发的都是新订阅，而响应始终为 "2"，条件 response.equals("1") 永不满足，takeUntil 无法截断流。

✅ 正确做法是：将轮询建模为“失败驱动的重试”而非“无条件重复”。即：仅当响应不符合预期时主动报错，再通过 retryWhen 控制重试次数与间隔。

Reactivation Specialist · 流失召回

适用于用户提出以下需求时：“构建赢回活动（win-back campaign）”、“重新激活已流失订阅者（re-engage lapsed subscribers）”、“执行重新授权 / 重新同意（re-permission / re-consent）清理”或“清理无效联系人列表（sunset my dead list）”；输出一个闭环式再激活方案——包括：流失用户群组（lapsed-cohort）定义、分阶段优惠阶梯（staged offer ladder）、重新授权（re-consent / re-p

下载

以下是推荐实现（适配 WebFlux 场景）：

import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;
import java.time.Duration;

// 假设 webClient.getResponse() 返回 Mono<String>
Mono<String> pollingMono = Mono.defer(() -> webClient.getResponse())
    // 若响应不是 "1"，则抛出异常（触发重试）
    .filter(response -> "1".equals(response))
    .switchIfEmpty(Mono.error(new RuntimeException("Expected '1', got: " + response)));

// 最多重试 4 次（即最多发起 5 次请求：第 1 次 + 4 次重试），每次间隔 500ms
Mono<String> result = pollingMono
    .retryWhen(Retry.fixedDelay(4, Duration.ofMillis(500))
        .doBeforeRetry(sig -> System.out.println("Retrying...")));

result.subscribe(
    response -> System.out.println("✅ Success! Response: " + response),
    error -> System.err.println("❌ Failed after all retries: " + error.getMessage())
);

? 关键要点总结：

• ✅ 使用 filter().switchIfEmpty(Mono.error()) 显式将“非目标响应”转化为错误，这是 retryWhen 起效的前提；
• ✅ Retry.fixedDelay(4, ...) 表示 最多重试 4 次（总请求次数 = 1 + 4 = 5），非“总共执行 4 次”；
• ✅ delayElements 属于数据流节流，适用于已发出元素的延时；而轮询本质是订阅时机控制，应使用 retryWhen 的延迟策略；
• ⚠️ 避免混用 repeatWhen/repeat 与 takeUntil/takeWhile —— 它们语义冲突，易引发不可预测行为；
• ? 如需严格限制总耗时（如超时退出），可在链末尾添加 .timeout(Duration.ofSeconds(3))。

通过将轮询语义对齐 Reactor 的错误传播与重试模型，即可获得精确可控、可终止、可观测的异步轮询逻辑。