rxjava-to-coroutines-migration

RxJava 到 Kotlin 协程迁移技能

一项专用技能，旨在安全且符合惯用写法地将 Android 或 Kotlin 代码库从 RxJava 重构为 Kotlin 协程和 Flow。

迁移映射指南

将 RxJava 组件迁移到 Kotlin 协程时，请使用以下标准映射：

1. 基础类型

• Single<T> -> suspend fun ...(): T
  • 一个异步值。
• Maybe<T> -> suspend fun ...(): T?
  • 一个可能不存在的异步值。
• Completable -> suspend fun ...()
  • 一个完成时不返回值的异步操作。
• Observable<T> -> Flow<T>
  • 一个冷值流。
• Flowable<T> -> Flow<T>
  • 协程 Flow 原生支持背压。

2. Subject 到热流

• PublishSubject<T> -> MutableSharedFlow<T>
  • 向多个订阅者广播事件。如需缓冲，请使用 MutableSharedFlow(extraBufferCapacity = ...)。
• BehaviorSubject<T> -> MutableStateFlow<T>
  • 保存状态，并向新订阅者发出当前/最新值。需要初始值。
• ReplaySubject<T> -> MutableSharedFlow<T>(replay = N)
  • 向新订阅者重放最近发出的 N 个值。

3. 调度器到调度器上下文

• Schedulers.io() -> Dispatchers.IO
• Schedulers.computation() -> Dispatchers.Default
• AndroidSchedulers.mainThread() -> Dispatchers.Main
• 上下文切换：通常使用 withContext(Dispatcher) 或 flowOn(Dispatcher)（适用于 Flow）替代 subscribeOn 和 observeOn。

4. 操作符

• map -> map
• filter -> filter
• flatMap -> flatMapMerge（并发）或 flatMapConcat（顺序）
• switchMap -> flatMapLatest
• doOnNext / doOnSuccess -> onEach
• onErrorReturn / onErrorResumeNext -> catch { emit(...) }
• startWith -> onStart { emit(...) }
• combineLatest -> combine
• zip -> zip
• delay -> delay（挂起函数）或 onEach { delay(...) }

5. 执行与生命周期

• subscribe() -> 在 CoroutineScope 中对 Flow 使用 collect {}，或直接调用挂起函数。
• Disposable.dispose() -> Job.cancel()
• CompositeDisposable.clear() -> 取消父级 CoroutineScope 或 Job。

执行步骤

1. 分析 RxJava 链：确定源类型（Single、Observable 等）、使用的操作符以及订阅发生的位置。
2. 转换源：首先修改仓库或数据源层的返回类型。一次性操作转换为 suspend 函数，数据流转换为 Flow。
3. 重写操作符：将 RxJava 操作符转换为对应的 Flow 或协程操作。请注意，许多 RxJava 操作符可直接替换为 map 或 onEach 代码块中的标准 Kotlin 集合/序列操作。
4. 更新订阅：在 ViewModel 或 Presenter 中，将 .subscribe(...) 替换为 launch { ... } 和 .collect { ... }。确保 launch 绑定到正确的生命周期作用域（例如 viewModelScope）。
5. 处理错误：用挂起函数周围的 try/catch 或 Flow 上的 .catch { } 操作符替换 onError 代码块。
6. 处理线程：移除 .subscribeOn() 和 .observeOn()。必要时使用 withContext，或使用 .flowOn() 更改上游 Flow 的上下文。

转换示例

RxJava：

fun getUser(id: String): Single<User> { ... }

disposable.add(
    getUser("123")
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe({ user ->
            view.showUser(user)
        }, { error ->
            view.showError(error)
        })
)

协程/Flow：

suspend fun getUser(id: String): User { ... } // Internally uses withContext(Dispatchers.IO) if needed

viewModelScope.launch {
    try {
        val user = getUser("123")
        view.showUser(user)
    } catch (e: Exception) {
        view.showError(e)
    }
}

最佳实践

• 优先使用挂起函数： 除非确实需要随时间产生多个值的数据流，否则默认使用 suspend 函数而非 Flow。Single 和 Completable 几乎总是转换为 suspend 函数。
• 状态处理： 在 ViewModel 中使用 StateFlow 向 UI 暴露状态，而不是使用 BehaviorSubject 或 LiveData。
• 生命周期感知： 在 UI 层收集 Flow 时使用 repeatOnLifecycle 或 flowWithLifecycle，避免视图不可见时仍在后台工作。