java - RxJava Multithreading with Realm - 从不正确的线程访问 Realm

Question

背景

我在我的应用程序中使用 Realm。加载数据后，它会进行密集处理，因此处理发生在后台线程上。

使用的编码模式是工作单元模式，Realm 仅存在于 DataManager 下的存储库中。这里的想法是每个存储库可以有不同的数据库/文件存储解决方案。

我尝试过的

下面是一些与我的 FooRespository 类中的类似代码的示例。

这里的想法是获取 Realm 的一个实例，用于查询 Realm 中感兴趣的对象，返回它们并关闭 Realm 实例。请注意，这是同步的，最后将对象从 Realm 复制到非托管状态。

public Observable<List<Foo>> getFoosById(List<String> fooIds) {

    Realm realm = Realm.getInstance(fooRealmConfiguration);

    RealmQuery<Foo> findFoosByIdQuery = realm.where(Foo.class);

    for(String id : fooIds) {

        findFoosByIdQuery.equalTo(Foo.FOO_ID_FIELD_NAME, id);
        findFoosByIdQuery.or();
    }

    return findFoosByIdQuery
            .findAll()
            .asObservable()
            .doOnUnsubscribe(realm::close)
            .filter(RealmResults::isLoaded)
            .flatMap(foos -> Observable.just(new ArrayList<>(realm.copyFromRealm(foos))));
}

此代码稍后通过 RxJava 与繁重的处理代码结合使用:

dataManager.getFoosById(foo)
            .flatMap(this::processtheFoosInALongRunningProcess)
            .subscribeOn(Schedulers.io()) //could be Schedulers.computation() etc
            .subscribe(tileChannelSubscriber);

阅读文档后，我相信上面的代码应该可以工作，因为它不是异步的，因此不需要循环线程。我在同一个线程中获取 Realm 的实例，因此它不会在线程之间传递，对象也不会。

问题

当执行上面的命令时，我得到

Realm access from incorrect thread. Realm objects can only be accessed on the thread they were created.

这似乎不对。我唯一能想到的是，Realm 实例池正在为我提供一个使用主线程从另一个进程创建的现有实例。

Answer 1

好吧

return findFoosByIdQuery
        .findAll()
        .asObservable()

这发生在 UI 线程上，因为这是您最初调用它的地方

.subscribeOn(Schedulers.io())

Aaaa 然后您在 Schedulers.io() 上修改它们。

不，那不是同一个线程!

尽管我不喜欢从零拷贝数据库中复制的方法，但由于滥用 realmResults.asObservable，您当前的方法充满了问题()，所以这里有一个关于你的代码应该是什么的剧透:

public Observable<List<Foo>> getFoosById(List<String> fooIds) {
   return Observable.defer(() -> {
       try(Realm realm = Realm.getInstance(fooRealmConfiguration)) { //try-finally also works
           RealmQuery<Foo> findFoosByIdQuery = realm.where(Foo.class);
           for(String id : fooIds) {
               findFoosByIdQuery.equalTo(FooFields.ID, id);
               findFoosByIdQuery.or(); // please guarantee this works?
           }
           RealmResults<Foo> results = findFoosByIdQuery.findAll();
           return Observable.just(realm.copyFromRealm(results));
       }
   }).subscribeOn(Schedulers.io());
}