Realm +改造+RxJava : Concat and SubscribeOn

Question

我在使用 RxJava concat 运算符时遇到问题。我有两个可观察对象，第一个从服务器数据库发出结果，另一个从本地数据库发出结果，然后我连接:

// Uses a Realm in the UI thread
Observable<MyResult> remoteObservable = mRemoteDataSource.find(tId);

// Uses Retrofit
Observable<MyResult> localObservable = mLocalDataSource.find(tId);

Observable.concat(localObservable, remoteObservable)
    .doOnNext(result -> /* Do my stuff */)
    .observeOn(AndroidSchedulers.mainThread())
    .doOnError(throwable -> throwable.printStackTrace())
    .subscribe()

所以这给我带来了问题，因为我没有使用 subscribeOn() 串联的可观察对象在 AndroidScheduler.MainThread() 上运行，并且这不会运行远程和它启动了一个NetworkOnMainThreadException。

如果我实现了 subscribeOn(Schedulers.computation())，我会从不正确的线程获得 Realm 访问权限。 Realm 对象只能在创建它们的线程上访问，因为 Observable 当然不是在 Realm 实例确实存在的线程上运行。

我搜索了其他问题，但没有得到任何有用的东西，我检查了realm制作的示例:https://github.com/realm/realm-java/blob/master/examples/rxJavaExample/src/main/java/io/realm/examples/rxjava/retrofit/RetrofitExample.java但奇怪的是，我看到改造的可观察对象没有订阅任何内容，而且它有效。

为什么它在示例中有效，而在我的代码中却不能执行相同的操作？有什么建议吗？

Answer 1

我相信您应该在正确的地方使用subscribeOn()。

// Uses a Realm in the UI thread
Observable<MyResult> realmObservable = mRealmDataSource.find(tId).subscribeOn(AndroidSchedulers.mainThread());

// Uses Retrofit
Observable<MyResult> retrofitObservable = mRetrofitDataSource.find(tId).subscribeOn(Subscribers.io());

Observable.concat(realmObservable, retrofitObservable)
    .doOnNext(result -> /* Do my stuff */)
    .subscribeOn(AndroidSchedulers.mainThread())
    .observeOn(AndroidSchedulers.mainThread())
    .doOnError(throwable -> throwable.printStackTrace())
    .subscribe()

看看它是否能解决您的问题。