RxJava 没有在后台线程上运行答案

【问题标题】：RxJava not running on Background threadRxJava 没有在后台线程上运行
【发布时间】：2018-06-18 05:33:56
【问题描述】：

我正在尝试在 Room 中保存数据，它需要一些后台线程来保存数据。所以我创建了一个这样的可观察对象

val obs: Observable<MutableLiveData<List<Source>>>? = Observable.fromCallable(object :Callable<MutableLiveData<List<Source>>>{
            override fun call(): MutableLiveData<List<Source>> {

                return mutableLiveData
            }
        })

那我就这样订阅、观察、退订

obs?.subscribeOn(Schedulers.io())?.observeOn(AndroidSchedulers.mainThread())?.unsubscribeOn(Schedulers.io())
                ?.subscribe(object : Observer<MutableLiveData<List<Source>>>{
                    override fun onComplete() {

                    }

                    override fun onSubscribe(d: Disposable?) {

                    }

                    override fun onNext(value: MutableLiveData<List<Source>>?) {
                        for(source in value!!.value!!.iterator()){
                            sourceDao.insert(source)//this is the line number 87, that logcat is pointing
                        }
                    }

                    override fun onError(e: Throwable?) {
                        e?.printStackTrace()
                    }
                })

我在 Schedulers.io 线程上订阅它，然后在 AndroidSchedulers.mainThread() 上观察它，但我仍然没有遇到后台线程错误。更具体的

    Caused by: java.lang.IllegalStateException: Cannot access database on the main thread since it may potentially lock the UI for a long period of time.
        at android.arch.persistence.room.RoomDatabase.assertNotMainThread(RoomDatabase.java:204)
        at android.arch.persistence.room.RoomDatabase.beginTransaction(RoomDatabase.java:251)
06-18 11:11:08.674 3732-3732/com.theanilpaudel.technewspro W/System.err:     at com.package.myapp.room.SourceDao_Impl.insert(SourceDao_Impl.java:63)
        at com.package.myapp.main.MainRepository$saveToRoom$1.onNext(MainRepository.kt:87)
        at com.package.myapp.main.MainRepository$saveToRoom$1.onNext(MainRepository.kt:76)

【问题讨论】：

由于您的observeOn 操作，onNext 在 MainThread 上执行。
我们还有哪些其他线程？

标签： android kotlin rx-java2 rx-android android-room

【解决方案1】：

在 Observable 中而不是在 Observer 中执行数据库操作：

val obs: Observable<MutableLiveData<List<Source>>>? = Observable.fromCallable(object :Callable<MutableLiveData<List<Source>>>{
        override fun call(): MutableLiveData<List<Source>> {
             for(source in mutableLiveData!!.value!!.iterator()){
                  sourceDao.insert(source)
             }
            return mutableLiveData
        })
        .subscribeOn(Schedulers.io())?.observeOn(AndroidSchedulers.mainThread())?.unsubscribeOn(Schedulers.io())
            ?.subscribe(object : Observer<MutableLiveData<List<Source>>>{
                override fun onComplete() {

                }

                override fun onSubscribe(d: Disposable?) {

                }

                override fun onNext(value: MutableLiveData<List<Source>>?) {
                    // Nothing todo right more
                }

                override fun onError(e: Throwable?) {
                    e?.printStackTrace()
                }
            })
    })

【讨论】：

我试过了，但它不起作用。不是一样的吗
不，因为插入是在 Observable 中而不是在 Observer 中执行的。不工作是没有有效的错误描述。

【解决方案2】：

这是有道理的，因为您有 observeOn 主线程并且您正在观察者中工作（observeOn 管理观察者的线程）。

要解决此问题，您可以在 obs 上使用 flatmap 并在其中执行 for 循环。由于 flatmap 要求你返回一个 Observable，你可以在你的 for 循环之后return Observable.just(true)。

【讨论】：