Kotlin中使用RxJava subscribe()遇类型不匹配错误及崩溃问题

我来一步步帮你拆解这些问题的原因和解决办法：

一、先搞懂「Type mismatch. Required: Disposable? Found: Unit」错误 #

说白了就是你试图把返回Unit的subscribe()调用结果，赋值给了Disposable?类型的变量，或者编译器误判了你调用的subscribe()重载，导致返回值类型不匹配。

为啥偏偏在RxBus的调用里触发？大概率是你用错了subscribe()的重载：

• 比如你误写了无参的subscribe()（这个重载确实返回Unit），却硬要把它赋值给Disposable变量，就会报错：

  // 错误示范：无参subscribe返回Unit，没法赋值给Disposable
  val disposable: Disposable? = RxBus.listen(DataEvent::class.java).subscribe()
  

• 也可能是你的lambda写法让编译器晕了头，选错了重载（比如同时传onNext和onError时语法写错）。

解决办法：用返回Disposable的subscribe()重载就行，比如只传处理事件的lambda：

val disposable = RxBus.listen(DataEvent::class.java).subscribe { event ->
    // 这里写处理DataEvent的逻辑
}

或者需要处理错误和完成事件的话，完整传参：

val disposable = RxBus.listen(DataEvent::class.java).subscribe(
    { event -> /* 收到事件时执行 */ },
    { error -> /* 出错时执行 */ },
    { /* 事件流结束时执行 */ }
)

二、搞定DisposableObserver配合subscribeWith()崩溃+只执行一次的问题 #

你说改用DisposableObserver后编译过了，但subscribeWith()只跑一次还崩溃，而且这个Observer是类的成员变量。这里大概率是两个原因：

1. Observer没做好生命周期管理，导致泄漏或提前失效 #

因为你的DisposableObserver是类成员，要是你的类（比如Activity/Fragment）销毁时没及时释放它，轻则内存泄漏，重则直接崩溃。另外，如果Observer被意外dispose了，后续的事件自然收不到。

解决办法：在类的生命周期销毁时（比如Activity的onDestroy、Fragment的onDestroyView），手动dispose这个Observer：

class YourActivity : AppCompatActivity() {
    private lateinit var eventObserver: DisposableObserver<DataEvent>

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        // 初始化Observer
        eventObserver = object : DisposableObserver<DataEvent>() {
            override fun onNext(event: DataEvent) { /* 处理事件 */ }
            override fun onError(e: Throwable) { /* 处理错误，别空着！空着容易崩 */ }
            override fun onComplete() { /* 处理事件流结束 */ }
        }
        // 绑定订阅
        RxBus.listen(DataEvent::class.java).subscribeWith(eventObserver)
    }

    override fun onDestroy() {
        super.onDestroy()
        // 销毁时释放资源
        if (!eventObserver.isDisposed) {
            eventObserver.dispose()
        }
    }
}

注意：onError方法别空实现，不然遇到错误时会直接抛出未捕获异常导致崩溃！

2. RxBus的实现有问题 #

如果你的RxBus是用PublishSubject实现的，它只会给订阅之后的事件发通知；要是你的Bus实现有bug（比如每次listen都新建一个Subject，或者Subject被意外调用了onComplete/dispose），就会导致只能收到一次事件。

检查你的RxBus实现，确保是单例且Subject没被意外终止：

object RxBus {
    // 用toSerialized保证线程安全
    private val bus = PublishSubject.create<Any>().toSerialized()

    fun post(event: Any) {
        bus.onNext(event)
    }

    // 泛型方法，自动过滤事件类型
    inline fun <reified T> listen(): Observable<T> {
        return bus.ofType(T::class.java)
    }
}

3. 别让Observable提前结束 #

如果你的RxBus的Subject被调用了onComplete，那整个事件流就终止了，后续再post事件也收不到。检查代码里有没有不小心调用bus.onComplete()的地方，把它删掉就行。

总结一下：

• 第一个类型不匹配错误，换个返回Disposable的subscribe()重载就搞定了；
• 第二个崩溃+只执行一次的问题，重点盯Observer的生命周期管理和RxBus的实现，记得在类销毁时释放Observer，同时保证Subject没被意外终止。

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