RxJava 2中创建可操控Emitter的Observable用于Android MVVM的正确方式

嘿，这个思路理论上是能跑通的，但在Android ViewModel的场景下，有不少需要注意的细节，而且还有更符合RxJava最佳实践的替代方案——咱们一步步说：

先给你的思路打个底：理论可行，但有坑 #

直接持有Emitter并手动调用onNext()来发送事件，本身是符合Observable的设计逻辑的，确实能实现“随时向View侧的订阅者传递数据”的需求。但在Android的生命周期环境里，这种写法很容易踩内存泄漏的坑：

• 如果View（比如Activity/Fragment）已经销毁，但ViewModel还存活（比如屏幕旋转导致重建），而Emitter又持有对订阅者的引用，就会导致View无法被GC回收，引发内存泄漏。
• 另外，手动持有Emitter还需要额外处理空值、订阅取消后的清理逻辑，稍不注意就会出现空指针或者无效事件发送的问题。

更推荐的实践：用Subject代替手动持有Emitter #

在RxJava里，Subject本身既是Observable又是Observer，完美适配ViewModel向View暴露数据流的场景，而且自带更安全的生命周期管理逻辑。根据你的需求，这里分两种情况推荐：

1. 传递状态类数据（比如当前选中的时间） #

如果需要让新订阅的View能立即拿到最新的时间状态，用BehaviorSubject最合适——它会保存最新的事件，新订阅者订阅时会先收到这个最新值：

class TimeViewModel : ViewModel() {
    // 内部用私有Subject负责发送事件
    private val _timeSelectedSubject = BehaviorSubject.createDefault<Long>(System.currentTimeMillis())
    
    // 对外暴露只读的Observable，防止View侧误发事件
    val timeSelectedObservable: Observable<Long> = _timeSelectedSubject

    // ViewModel内部触发事件的方法
    fun onUserSelectedTime(time: Long) {
        _timeSelectedSubject.onNext(time)
    }
}

2. 传递一次性事件（比如操作成功提示） #

如果是不需要保留状态的一次性事件，用PublishSubject就够了——它只会给订阅后的发送的事件：

class TimeViewModel : ViewModel() {
    private val _timeEventSubject = PublishSubject.create<TimeEvent>()
    val timeEventObservable: Observable<TimeEvent> = _timeEventSubject

    fun showTimeToast(time: Long) {
        _timeEventSubject.onNext(TimeEvent.ShowToast("选中时间：$time"))
    }

    sealed class TimeEvent {
        data class ShowToast(val message: String) : TimeEvent()
    }
}

不管用哪种Subject，View侧都要记得在销毁时清理订阅，比如用CompositeDisposable：

// 在Activity/Fragment里
private val compositeDisposable = CompositeDisposable()

override fun onCreate(savedInstanceState: Bundle?) {
    super.onCreate(savedInstanceState)
    viewModel.timeSelectedObservable
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe { selectedTime ->
            // 更新UI逻辑
        }
        .addTo(compositeDisposable)
}

override fun onDestroy() {
    super.onDestroy()
    compositeDisposable.clear()
}

如果非要坚持用Emitter的话，一定要做好这些 #

如果你确实需要用Observable.create()来获取Emitter，那必须严格处理以下两点：

1. 在ViewModel的onCleared()方法里调用Emitter.onComplete()，并清空引用，避免内存泄漏；
2. 给Emitter设置Cancellable，在订阅取消时自动清理引用。

示例代码：

class TimeViewModel : ViewModel() {
    private var timeEmitter: Emitter<Long>? = null
    val timeSelectedObservable: Observable<Long> = Observable.create { emitter ->
        timeEmitter = emitter
        // 设置订阅取消时的清理逻辑
        emitter.setCancellable {
            timeEmitter = null
        }
    }

    fun sendTimeEvent(time: Long) {
        timeEmitter?.onNext(time)
    }

    override fun onCleared() {
        super.onCleared()
        timeEmitter?.onComplete()
        timeEmitter = null
    }
}

但还是那句话，这种写法比Subject繁琐太多，出错概率也高，除非有特殊场景需求，否则不推荐。

总结 #

你的核心思路是对的，但在Android ViewModel的场景下，用Subject系列实现会更简洁、安全，也更符合RxJava的最佳实践。关键是要区分状态和事件的场景，选对合适的Subject，同时一定要管理好订阅的生命周期，避免内存泄漏。

内容的提问来源于stack exchange，提问作者A. Steenbergen