RxJava调用syncUser重复添加观察者致API请求递增问题排查

判断结论 #

你的判断完全正确。Room数据库查询返回的Flowable属于冷流，每次调用subscribe()都会创建新的订阅实例，若之前的订阅未被销毁，这些重复的订阅会同时存在。当数据流触发时，所有活跃的订阅都会执行subscribe块内的loadUserOptions()，导致API请求次数随调用次数递增。

修复方案 #

1. 手动管理订阅生命周期，避免重复订阅 #

用Disposable对象保存订阅实例，每次调用syncUser()前先取消旧订阅，再创建新订阅：

private var userSubscription: Disposable? = null

fun syncUser() {
    // 取消之前未销毁的订阅
    userSubscription?.dispose()
    
    userSubscription = getUserUseCase.getUser()
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe({ user ->
            loadUserOptions()
        }, { throwable ->
            // 处理错误逻辑
        })
}

注意：在组件（如Activity/Fragment/ViewModel）生命周期结束时，必须调用userSubscription?.dispose()，防止内存泄漏。

2. 改用单次订阅类型（Single/Maybe） #

如果syncUser()只需要获取一次用户数据，不需要持续监听数据库变化，将getUserUseCase.getUser()的返回类型改为Single<User>或Maybe<User>：

// 先修改getUser()的返回类型为Single<User>
fun syncUser() {
    getUserUseCase.getUser()
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe({ user ->
            loadUserOptions()
        }, { throwable ->
            // 错误处理
        })
}

Single会确保只执行一次订阅逻辑，完成后自动销毁订阅，不会产生重复触发的问题。

3. 转换为热流共享订阅（适用于需要持续监听的场景） #

如果需要持续监听用户数据变化，但不想重复执行查询，可将冷流转换为热流，让所有订阅共享同一个数据流：

// 初始化时将冷流转为热流
private val sharedUserFlowable = getUserUseCase.getUser()
    .publish()
    .refCount() // 自动管理订阅，当无活跃订阅时销毁流

fun syncUser() {
    sharedUserFlowable
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe({ user ->
            loadUserOptions()
        }, { throwable ->
            // 错误处理
        })
}

publish().refCount()会让多个订阅共享同一个数据流实例，避免重复执行Room查询。

4. 迁移到Kotlin协程的StateFlow/SharedFlow（推荐） #

若项目已使用Kotlin协程，可将Room查询返回的Flowable转为Flow，再用StateFlow实现共享订阅：

import kotlinx.coroutines.flow.firstOrNull
import kotlinx.coroutines.flow.stateIn
import kotlinx.coroutines.launch

// 在ViewModel中初始化StateFlow
private val userStateFlow = getUserUseCase.getUser() // 假设返回Flow<User>
    .stateIn(
        scope = viewModelScope,
        started = SharingStarted.WhileSubscribed(5000), // 闲置5秒后自动取消订阅
        initialValue = null
    )

fun syncUser() {
    viewModelScope.launch {
        // 仅收集一次当前用户数据
        userStateFlow.firstOrNull()?.let {
            loadUserOptions()
        }
    }
}

StateFlow会自动管理订阅生命周期，多次调用syncUser()也不会创建重复订阅，同时避免内存泄漏。

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