reactor.util.retry.Retry是否有重试成功触发方法？替代方案有哪些

Reactor 官方的 reactor.util.retry.Retry 确实没有内置 doOnRetrySuccess 类的回调，你可以通过状态标记配合现有API实现需求，以下是两种生产可用的实现方式：

方案1：简单场景用原子变量标记（单订阅/低并发场景） #

适合单实例订阅、不需要多订阅隔离的场景，实现最简单：

import reactor.util.retry.Retry;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicBoolean;

// 标记是否发生过重试
AtomicBoolean hasRetried = new AtomicBoolean(false);

// 你的SSE原始流
yourSseFlux
    // 发生错误时标记后续是重试请求
    .doOnError(err -> hasRetried.set(true))
    // 你的现有重试规则
    .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
        .filter(err -> isRetryableError(err)) // 替换为你自己的重试条件判断
    )
    .doOnNext(item -> {
        // 是重试成功后返回的第一个数据，执行自定义操作
        if (hasRetried.compareAndSet(true, false)) {
            // 替换为你要执行的重试成功逻辑
            runYourRetrySuccessAction();
        }
        // 正常的业务处理逻辑
        processNormalSseData(item);
    })
    .subscribe();

方案2：上下文状态标记（多订阅/生产级场景） #

适合需要多订阅隔离、严格符合响应式规范的场景，避免多线程下状态串扰：

import reactor.util.context.ContextView;
import reactor.util.retry.Retry;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicBoolean;

// 定义全局唯一的上下文Key
private static final ContextView.Key<AtomicBoolean> RETRY_STATE_KEY = ContextView.key("retry_has_occurred");

yourSseFlux
    .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
        .filter(err -> isRetryableError(err))
        // 每次重试前更新上下文的重试标记
        .doBeforeRetry(signal -> signal.contextView().get(RETRY_STATE_KEY).set(true))
    )
    .doOnNext(item -> {
        AtomicBoolean retryState = currentContext().get(RETRY_STATE_KEY);
        if (retryState.compareAndSet(true, false)) {
            runYourRetrySuccessAction();
        }
        processNormalSseData(item);
    })
    // 每个订阅初始化独立的重试状态
    .contextWrite(ctx -> ctx.put(RETRY_STATE_KEY, new AtomicBoolean(false)))
    .subscribe();

实现说明 #

两种方案的核心逻辑都是标记重试事件的发生，在流成功恢复后检查标记触发自定义操作，区别仅在于状态存储的方式适配不同的使用场景。如果你的重试逻辑是针对单次连接的SSE流，方案1足够满足需求，实现成本更低。

内容的提问来源于stack exchange，提问作者Arun Kumar V