當在Schedulers.io上訂閱Observable時，從不正確的線程訪問領域

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) { 
        } 
       }); 

這是我觀察到的地方我做的境界交易

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

這是因爲你與observeOn(AndroidSchedulers.mainThread())結合.subscribeOn(Schedulers.IO()。

這意味着您執行io()線程的工作，但嘗試在主線程上使用結果。由於我們使用線程約束對象，所以這與當前的Realms線程策略不符。我們在此處支持此用例的問題如下：https://github.com/realm/realm-java/issues/1208

在此之前，您必須將Realm數據複製到標準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()); 

    } 
});