RxJava实现OAuth2刷新令牌：请求等待令牌刷新完成方案咨询

嘿，我来帮你捋清楚这个问题！在RxJava环境里处理OAuth2的令牌刷新，核心就是要让令牌刷新操作和后续的请求重试串联起来，同时还要避免多个并发请求重复触发刷新接口，这也是很多新手容易卡壳的点。

核心思路 #

当OkHttp的Authenticator捕获到401未授权响应时，我们需要：

1. 先判断是否已经在刷新令牌了，避免并发请求重复调用刷新接口
2. 发起异步的令牌刷新请求（用RxJava），等待请求完成拿到新令牌
3. 用新令牌重新构建原请求并返回，让OkHttp自动重试

改造后的OAuthAuthenticator示例代码 #

class OAuthAuthenticator(
    private val authStateManager: AuthStateManager,
    private val authApi: AuthApi // 你的刷新令牌API服务
) : Authenticator {

    // 用原子布尔值标记是否正在刷新，防止并发刷新
    private val isRefreshing = AtomicBoolean(false)
    // 同步锁，保证刷新操作的原子性
    private val refreshLock = Any()

    override fun authenticate(route: Route?, response: Response): Request? {
        synchronized(refreshLock) {
            val currentAuthState = authStateManager.getCurrentAuthState()
            val currentAccessToken = currentAuthState.accessToken

            // 避免无限循环：如果当前请求已经携带了当前令牌还返回401，说明令牌确实失效
            if (response.request.header("Authorization") == "Bearer $currentAccessToken") {
                return try {
                    if (isRefreshing.get()) {
                        // 已有请求在刷新，等待刷新完成后重试
                        waitForRefreshAndRetry(response)
                    } else {
                        isRefreshing.set(true)
                        // 用RxJava发起刷新请求，blockingGet等待异步结果（OkHttp在后台线程调用此方法，不会卡主线程）
                        val newAuthState = authApi.refreshToken(currentAuthState.refreshToken)
                            .subscribeOn(Schedulers.io())
                            .blockingGet()

                        // 保存新的令牌状态到本地
                        authStateManager.saveAuthState(newAuthState)

                        // 用新令牌重新构建请求
                        response.request.newBuilder()
                            .header("Authorization", "Bearer ${newAuthState.accessToken}")
                            .build()
                    }
                } catch (e: Exception) {
                    // 刷新失败（比如刷新令牌过期），清除本地状态并引导登录
                    authStateManager.clearAuthState()
                    // 这里可以用EventBus/ViewModel通知UI跳转登录页
                    null
                } finally {
                    isRefreshing.set(false)
                }
            }

            // 如果请求没带令牌，直接用当前令牌重试
            return response.request.newBuilder()
                .header("Authorization", "Bearer $currentAccessToken")
                .build()
        }
    }

    private fun waitForRefreshAndRetry(response: Response): Request? {
        return try {
            // 循环等待刷新完成，直到isRefreshing变为false
            while (isRefreshing.get()) {
                Thread.sleep(100)
            }
            val newAccessToken = authStateManager.getCurrentAuthState().accessToken
            response.request.newBuilder()
                .header("Authorization", "Bearer $newAccessToken")
                .build()
        } catch (e: InterruptedException) {
            Thread.currentThread().interrupt()
            null
        }
    }
}

几个关键注意事项 #

• 并发控制：用AtomicBoolean和synchronized锁确保同一时间只有一个刷新请求在执行，其他请求等待刷新完成后复用新令牌，避免重复调用刷新接口浪费资源。
• RxJava的blockingGet()：因为OkHttp的authenticate是同步方法，所以用blockingGet()等待异步刷新请求完成是安全的——OkHttp会在后台线程调用这个方法，不会阻塞主线程。
• 防止无限循环：一定要检查当前请求是否已经携带了最新的令牌，否则会陷入“请求401→刷新→重试→又401”的死循环。
• 失败处理：如果刷新令牌本身也过期了，要及时清除本地的认证状态，引导用户重新登录，不要让请求一直重试。

更优雅的RxJava订阅方式（可选） #

如果你不想用blockingGet()，可以用PublishSubject来监听刷新状态，让等待的请求订阅这个Subject获取新令牌：

private val refreshSubject = PublishSubject.create<AuthState>()

// 在刷新逻辑里替换成：
authApi.refreshToken(currentAuthState.refreshToken)
    .subscribeOn(Schedulers.io())
    .subscribe(
        { newState ->
            authStateManager.saveAuthState(newState)
            refreshSubject.onNext(newState)
            refreshSubject.onComplete()
        },
        { error ->
            refreshSubject.onError(error)
        }
    )

// 等待刷新的请求用这个方式获取新令牌：
return refreshSubject
    .take(1)
    .map { newState ->
        response.request.newBuilder()
            .header("Authorization", "Bearer ${newState.accessToken}")
            .build()
    }
    .blockingGet()

本质上还是用blockingGet()来同步获取Rx流的结果，因为Authenticator必须返回一个Request对象，没法异步返回。

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