java.lang.IllegalStateException：从不正确的线程访问域。领域对象只能在创建它们的线程上访问

Question

我想通过将使用领域的改造observable返回的服务调用的响应保存到本地数据库，试图将改造2与RxJava和领域相结合。所以我得到一个异常说Realm从不正确的线程访问。 这里是我的代码：

径1：

restApi.userEntityList() 
      .map(userEntityDataMapper::transformAllToRealm) 
      .doOnNext(userRealmModels -> { 
       if (userRealmModels != null){ 
        mRealm = Realm.getInstance(mContext); 
        mRealm.asObservable() 
          .map(realm -> mRealm.copyToRealmOrUpdate(userEntity)) 
          .subscribe(new Subscriber<Object>() { 
           @Override 
           public void onCompleted() { 

           } 

           @Override 
           public void onError(Throwable e) { 
            e.printStackTrace(); 
           } 

           @Override 
           public void onNext(Object o) { 
            Log.d("RealmManager", "user added!"); 
           } 
          }); 
       }}) 
      .map(userEntityDataMapper::transformAll) 
      .subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(new Subscriber<List<User>>() { 
       @Override 
       public void onCompleted() { 
        hideViewLoading(); 
       } 

       @Override 
       public void onError(Throwable e) { 
        hideViewLoading(); 
        showErrorMessage(new DefaultErrorBundle((Exception) e)); 
        showViewRetry(); 
       } 

       @Override 
       public void onNext(List<User> users) { 
        showUsersCollectionInView(users); 
       } 
      }); 

开拓者2：

restApi.userEntityList() 
      .map(userEntityDataMapper::transformAllToRealm) 
      .doOnNext(userRealmModels -> { 
       if (userRealmModels != null) { 
        mRealm = Realm.getInstance(mContext); 
        mRealm.beginTransaction(); 
        mRealm.copyToRealmOrUpdate(userEntity); 
        mRealm.commitTransaction(); 
       } 
      }) 
      .map(userEntityDataMapper::transformAll) 
      .subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(new Subscriber<List<User>>() { 
       @Override 
       public void onCompleted() { 
        hideViewLoading(); 
       } 

       @Override 
       public void onError(Throwable e) { 
        hideViewLoading(); 
        showErrorMessage(new DefaultErrorBundle((Exception) e)); 
        showViewRetry(); 
       } 

       @Override 
       public void onNext(List<User> users) { 
        showUsersCollectionInView(users); 
       } 
      }); 

的logcat：

W/System.err: java.lang.IllegalStateException: Realm access from incorrect thread. Realm objects can only be accessed on the thread they were created. 
at io.realm.BaseRealm.checkIfValid(BaseRealm.java:349) 
at io.realm.BaseRealm.commitTransaction(BaseRealm.java:291) 
at io.realm.Realm.commitTransaction(Realm.java:108) 
at com.zeyad.cleanarchitecturet.data.db.RealmManagerImpl.put(RealmManagerImpl.java:66) 
at com.zeyad.cleanarchitecturet.data.db.RealmManagerImpl.putAll(RealmManagerImpl.java:91) 
at com.zeyad.cleanarchitecturet.data.repository.datasource.CloudUserDataStore$2.call(CloudUserDataStore.java:36) 
at com.zeyad.cleanarchitecturet.data.repository.datasource.CloudUserDataStore$2.call(CloudUserDataStore.java:32) 
at rx.Observable$11.onNext(Observable.java:4445) 
at rx.internal.operators.OperatorDoOnEach$1.onNext(OperatorDoOnEach.java:80) 
at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:54) 
at rx.internal.operators.OperatorMerge$MergeSubscriber.emitScalar(OperatorMerge.java:477) 
at rx.internal.operators.OperatorMerge$MergeSubscriber.tryEmit(OperatorMerge.java:435) 
at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:228) 
at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:142) 
at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:54) 
at retrofit.RxJavaCallAdapterFactory$CallOnSubscribe.call(RxJavaCallAdapterFactory.java:113) 
at retrofit.RxJavaCallAdapterFactory$CallOnSubscribe.call(RxJavaCallAdapterFactory.java:88) 
at rx.Observable$2.call(Observable.java:162) 
at rx.Observable$2.call(Observable.java:154) 
at rx.Observable$2.call(Observable.java:162) 
at rx.Observable$2.call(Observable.java:154) 
at rx.Observable$2.call(Observable.java:162) 
at rx.Observable$2.call(Observable.java:154) 
at rx.Observable$2.call(Observable.java:162) 
at rx.Observable$2.call(Observable.java:154) 
at rx.Observable$2.call(Observable.java:162) 
at rx.Observable$2.call(Observable.java:154) 
at rx.Observable$2.call(Observable.java:162) 
at rx.Observable$2.call(Observable.java:154) 
at rx.Observable$2.call(Observable.java:162) 
at rx.Observable$2.call(Observable.java:154) 
at rx.Observable.unsafeSubscribe(Observable.java:8098) 
at rx.internal.operators.OperatorSubscribeOn$1$1.call(OperatorSubscribeOn.java:62) 
at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55) 
at rx.schedulers.ExecutorScheduler$ExecutorSchedulerWorker.run(ExecutorScheduler.java:98) 
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1113) 
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:588) 
at java.lang.Thread.run(Thread.java:818) 
Caused by: rx.exceptions.OnErrorThrowable$OnNextValue: OnError while emitting onNext value: java.util.ArrayList.class 
at rx.exceptions.Exceptions.throwOrReport(Exceptions.java:187) 
at rx.internal.operators.OperatorDoOnEach$1.onNext(OperatorDoOnEach.java:82) 
... 29 more 

Answer 1

我发现了如何解决它。要将服务器响应从改进2保存到领域，然后传递到用户界面。

Retrofit.Builder() 
      .baseUrl(RestApi.API_BASE_URL) 
      .client(okHttpClient) 
      .callbackExecutor(new JobExecutor()) 
      .addConverterFactory(GsonConverterFactory.create(new GsonBuilder() 
        .setExclusionStrategies(new ExclusionStrategy() { 
         @Override 
         public boolean shouldSkipField(FieldAttributes f) { 
          return f.getDeclaringClass().equals(RealmObject.class); 
         } 

         @Override 
         public boolean shouldSkipClass(Class<?> clazz) { 
          return false; 
         } 
        }).create())) 
      .addCallAdapterFactory(RxJavaCallAdapterFactory.create()) 
      .build(); 

然后：

restApi.userRealmList() 
    .doOnNext(userRealmModels -> { 
     if (userRealmModels != null) { 
      Realm realm = Realm.getInstance(mContext); 
      realm.beginTransaction(); 
      realm.copyToRealmOrUpdate(userRealmModels); 
      realm.commitTransaction(); 
      realm.close(); 
     }}) 
    .map(userEntityDataMapper::transformAll) 
    .subscribeOn(Schedulers.io()) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .subscribe(new Subscriber<List<User>>() { 
     @Override 
     public void onCompleted() { 
      hideViewLoading(); 
     } 

     @Override 
     public void onError(Throwable e) { 
      hideViewLoading(); 
      showErrorMessage(new DefaultErrorBundle((Exception) e)); 
      showViewRetry(); 
     } 

     @Override 
     public void onNext(List<User> users) { 
      showUsersCollectionInView(users); 
     } 
    }); 

Answer 2

我想你已经使用主线程创建的域对象和您正在另一个或后台线程中使用它。如果是这样，那么在创建它们的同一个线程中使用它们。 即

((Activity)mContext).runOnUiThread(new Runnable() { 
    @Override 
    public void run() { 
     // You code of realm goes here 
     if (userRealmModels != null) { 
       mRealm = Realm.getInstance(mContext); 
       mRealm.beginTransaction(); 
       mRealm.copyToRealmOrUpdate(userEntity); 
       mRealm.commitTransaction(); 
     } 
    } 
});