RxJava中Observable请求token过期后等待刷新token流并重订阅如何实现

核心实现思路基于retryWhen操作符做定制改造，你之前遇到重试3次的问题，本质是没有对错误类型做过滤、也未做刷新状态互斥控制，调整后即可实现你要的链式调用效果。

实现步骤 #

• 仅针对Token过期类错误触发重试逻辑，其他业务错误/网络错误直接抛给下游不重试
• 用原子布尔值维护刷新状态，避免多个请求同时触发多次Token刷新
• 刷新Token的Observable添加share()操作符，确保多请求等待时仅执行一次刷新动作

完整代码示例 #

首先封装可复用的重试逻辑转换器：

import java.util.concurrent.atomic.AtomicBoolean;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableTransformer;

// 全局维护的刷新状态，也可以放到单例的网络管理类中统一管理
private final AtomicBoolean isTokenRefreshing = new AtomicBoolean(false);

// 刷新Token的Observable，share()确保多订阅共享同一次执行结果
private final Observable<Token> refreshTokenObservable = yourRefreshTokenApi()
        .doOnNext(newToken -> {
            // 刷新成功后更新本地缓存的Token
            saveNewTokenToLocal(newToken);
            isTokenRefreshing.set(false);
        })
        .doOnError(e -> {
            isTokenRefreshing.set(false);
            // 可选：刷新Token失败直接跳转登录页
            navigateToLogin();
        })
        .share();

// 封装为Transformer 可直接给所有请求复用
public <T> ObservableTransformer<T, T> autoRetryAfterTokenRefresh() {
    return upstream -> upstream.retryWhen(errors ->
            errors.flatMap(throwable -> {
                // 只处理Token过期的错误，其他错误不重试
                if (!(throwable instanceof TokenExpiredException)) {
                    return Observable.error(throwable);
                }
                // CAS修改状态，只有第一个触发过期的请求会走到刷新逻辑
                if (isTokenRefreshing.compareAndSet(false, true)) {
                    return refreshTokenObservable;
                } else {
                    // 其他请求直接等待正在执行的刷新结果即可
                    return refreshTokenObservable;
                }
            })
    );
}

使用方式 #

完全符合你要的链式调用写法：

getFilms()
    .compose(autoRetryAfterTokenRefresh())
    .subscribe(films -> {
        // 处理接口返回的影片数据
    }, throwable -> {
        // 处理非Token过期的其他错误、或是Token刷新失败的异常
    });

常见问题说明 #

你之前使用retryWhen出现重复触发3次的问题，通常是以下两种原因导致：

1. 没有对错误类型做过滤，所有类型的报错都会触发重试逻辑
2. 没有加刷新状态控制，多个并发请求同时过期时会触发多次刷新请求
   使用上述方案后，只会在Token过期时触发1次刷新，刷新成功后自动重试1次原请求，完全符合需求。

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