RxJava中如何避免subscribe嵌套 借助map/flatMap实现单次订阅

现有代码的核心问题 #

原实现的嵌套订阅结构存在几个典型问题：

• 流处理节点内夹带着副作用逻辑（弹窗、按钮状态修改）
• 在map回调内部直接嵌套发起新的UseCase订阅，破坏了RxJava“单向链式流”的结构
• 输入快速变化时，旧的手续费查询请求不会被自动取消，容易出现新输入对应的手续费被旧响应覆盖的竞态问题
• 多层订阅需要单独做生命周期绑定，容易造成内存泄漏

重构实现逻辑 #

完全取消嵌套订阅的核心思路是：把所有分支的逻辑从“回调内发起新订阅”，改成用流转换操作符把不同分支的逻辑封装成可拼接的响应式流，最终整个链路只保留一次终点订阅。
关键实现要点：

• 用switchMap替代普通map+内部订阅的结构：每个输入分支返回对应的流（同步分支直接返回单值流，异步的手续费查询分支直接返回UseCase的原始流），由switchMap自动完成流的拼接，同时自动取消上一次未完成的异步请求
• 把所有UI副作用收敛到最终的订阅回调里统一处理，不要在流的中间转换节点执行UI操作
• 对输入做类型分层，避免直接在when分支里混杂不同类型的返回值，保证流的类型安全

重构后代码示例 #

// 提前定义密封类做状态/动作的类型约束，避免流的类型混乱
sealed class AmountInputState {
    object Empty : AmountInputState()
    object ExceedMaxLimit : AmountInputState()
    data class LegalValid(val amountFen: Int) : AmountInputState()
    object Invalid : AmountInputState()
}

sealed class UiAction {
    object DisableNextBtn : UiAction()
    object ShowMaxAmountError : UiAction()
    data class UpdateCommission(val content: String) : UiAction()
}

amount.textChanges.observable
    // 第一阶段：把原始文本输入转换为统一的输入状态
    .map { rawText ->
        val amount = rawText.toBigDecimalOrNull()
        when {
            rawText.isEmpty() -> AmountInputState.Empty
            amount == null -> AmountInputState.Invalid
            amount > BigDecimal(1500) -> AmountInputState.ExceedMaxLimit
            amount > BigDecimal.ZERO -> AmountInputState.LegalValid((amount * 100).toInt())
            else -> AmountInputState.Invalid
        }
    }
    .distinctUntilChanged()
    // 第二阶段：把输入状态转换为需要执行的UI动作流，自动拼接异步请求
    .switchMap { inputState ->
        when (inputState) {
            AmountInputState.Empty, AmountInputState.Invalid -> 
                Observable.just(UiAction.DisableNextBtn)
            AmountInputState.ExceedMaxLimit -> 
                Observable.just(UiAction.ShowMaxAmountError)
            is AmountInputState.LegalValid -> {
                getCommissionUseCase.execute(inputState.amountFen)
                    .map { payment ->
                        UiAction.UpdateCommission((payment.data.toDouble() / 100).toString())
                    }
                    // 异步请求加错误兜底，避免单次请求失败导致整个输入监听流中断
                    .onErrorReturnItem(UiAction.UpdateCommission("--"))
            }
        }
    }
    .untilDestroy()
    // 全链路仅保留这一次订阅，统一处理所有UI动作
    .subscribe { action ->
        when (action) {
            UiAction.DisableNextBtn -> nextBtnEnabled.consumer.accept(false)
            UiAction.ShowMaxAmountError -> showDialog(R.string.max_amount_error)
            is UiAction.UpdateCommission -> commission.accept(action.content)
        }
    }

实现收益：

• 全程仅保留最终点的一次订阅，不存在嵌套发起的子订阅，生命周期绑定只需要做一次，不会出现泄漏
• switchMap自动取消上一次未完成的手续费查询，从机制上解决快速输入时的竞态问题
• 流的分层清晰：文本输入→输入状态判断→动作生成→UI处理，每个节点职责单一，后续改逻辑不需要穿透嵌套回调

内容的提问来源于stack exchange，提问作者Dato.Beriashvili