当 Observable 在 Schedulers.io() 上订阅时从不正确的线程访问领域

Question

我订阅了 IO 调度程序。

getObservableItems(itModel).subscribeOn(Schedulers.io()).
                onBackpressureBuffer().
                observeOn(AndroidSchedulers.mainThread()).
                subscribe(new Observer<List<ItemModel>>() {
                    @Override
                    public void onCompleted() {
                    }

                    @Override
                    public void onError(Throwable e) {
                    }

                    @Override
                    public void onNext(List<ItemModel> resultItemModel) {
                    }
                });

这是我在进行 Realm 交易时的可观察对象

private Observable<List<ItemModel>> getObservableItems(ItModel itModel) {
        return Observable.create(subscriber -> {
            realm = Realm.getInstance(mContext);
            if (itModel != null) {
                ArrayList<String> ids = ProjectUtil.getId(itModel.getRequestUrl());
                DatabaseHelper.saveItemCategory(realm, itModel, ids.get(0), ids.get(1));
            }
            RealmQuery<ItemModel> itemModelRealmQuery = realm.where(ItemModel.class);
            /* Error on below line */
            resultItemModel = itemModelRealmQuery.equalTo("res_id", subCategoryModel.getId()).
                    equalTo("menu_grp_id", subCategoryModel.getMenu_grp_id()).findAll();
            subscriber.onNext(resultItemModel);
            subscriber.onCompleted();
        });
    }

Answer 1

这是因为您将 .subscribeOn(Schedulers.IO() 与 observeOn(AndroidSchedulers.mainThread()) 组合在一起。

这意味着您在 io() 线程上执行工作，但尝试在主线程上使用结果。这目前违反了 Realms 线程策略，因为我们使用线程限制对象。我们在此处支持此用例时遇到问题：https://github.com/realm/realm-java/issues/1208

在那之前，您要么必须将领域数据复制到标准 Java 对象，要么在同一线程上执行所有领域操作（通过删除 subscribeOn/observeOn）。

Answer 2

更新0.87.1后，Realm提供RxJava支持。

realm.where(ItemModel.class).findAllAsync().asObservable()
 .observeOn(AndroidSchedulers.mainThread())
 .subscribe(new Observer < RealmResults < ItemModel >> () {
  @Override
  public void onCompleted() {

  }

  @Override
  public void onError(Throwable e) {

  }

  @Override
  public void onNext(RealmResults < ItemModel > itemModels) {
   System.out.println(itemModels.get(0).getItem());

  }
 });

当 Observable 在 Schedulers.io() 上订阅时从不正确的线程访问领域

Realm access from incorrect thread when Observable subscribed on Schedulers.io()

android

realm

rx-android