Realm+Retrofit+RxJava:Concat和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()连接的observable是在AndroidScheduler.mainthread()上运行的，这不运行远程，它启动了NetworkOnMainThreadException。

如果我实现了订阅(scheduleers.computation())，我将从不正确的线程获得领域访问。领域对象只能在创建它们的线程上访问，因为Observable当然不在存在领域实例的线程上运行。

我已经搜索了其他问题，但没有得到任何有用的信息，我检查了由realm创建的示例：https://github.com/realm/realm-java/blob/master/examples/rxjavaexample/src/main/java/io/realm/examples/rxjava/retrofit/retrofitexample。java但是奇怪的是，我发现refitfit observable没有订阅任何内容，而且它可以工作。

为什么它在示例中起作用，而在我的代码中却不能这样做？有什么建议吗？

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()

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