Rxjava2只是方法 - 如何在另一个线程上运行房间插入？

Question

我有一个房间持续性数据库插入方法，它看起来是这样的：

@Dao 
public interface CountriesDao{ 

    @Insert(onConflict = REPLACE) 
    List<Long> addCountries(List<CountryModel> countryModel); 
} 

我知道这不能在主线程上运行。这里是我如何定义我的数据库：

Room.inMemoryDatabaseBuilder(context.getApplicationContext(), MyDatabase.class).build(); 

我想使用rxjava2，以便我不在mainthread上运行。我已经创建了以下方法：

public void storeCountries(List<CountryModel> countriesList) { 
     Observable.just(db.countriesDao().addCountries(countriesList)) 
       .subscribeOn(Schedulers.io()) 
       .observeOn(AndroidSchedulers.mainThread()) 
       .subscribe(new DefaultSubscriber<List<Long>>(){ 
      @Override 
      public void onSubscribe(@NonNull Disposable d) { 
       super.onSubscribe(d); 
      } 

      @Override 
      public void onNext(@NonNull List<Long> longs) { 
       super.onNext(longs); 
       Timber.d("insert countries transaction complete"); 
      } 

      @Override 
      public void onError(@NonNull Throwable e) { 
       super.onError(e); 
       Timber.d("error storing countries in db"+e); 
      } 

      @Override 
      public void onComplete() { 
       Timber.d("insert countries transaction complete"); 
      } 
     }); 
    } 

对于我来说，这显然现在正在另一个线程上运行。不是主线程，但是当我运行此代码时，出现以下错误：

由于：java.lang.IllegalStateException：无法访问主线程上的数据库，因为它可能会长时间锁定UI 。

完整的堆栈跟踪如下。为什么发生这种情况？

Process: com.mobile.myapp.staging, PID: 12990 
                      java.lang.IllegalStateException: Fatal Exception thrown on Scheduler. 
                       at io.reactivex.android.schedulers.HandlerScheduler$ScheduledRunnable.run(HandlerScheduler.java:111) 
                       at android.os.Handler.handleCallback(Handler.java:751) 
                       at android.os.Handler.dispatchMessage(Handler.java:95) 
                       at android.os.Looper.loop(Looper.java:154) 
                       at android.app.ActivityThread.main(ActivityThread.java:6077) 
                       at java.lang.reflect.Method.invoke(Native Method) 
                       at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:866) 
                       at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:756) 
                      Caused by: java.lang.IllegalStateException: Cannot access database on the main thread since it may potentially lock the UI for a long period of time. 
                       at android.arch.persistence.room.RoomDatabase.assertNotMainThread(RoomDatabase.java:138) 
                       at android.arch.persistence.room.RoomDatabase.beginTransaction(RoomDatabase.java:185) 
                       at com.mobile.myapp.data.room.dao.CountriesDao_Impl.addCountries(CountriesDao_Impl.java:165) 
                       at com.mobile.myapp.data.repositories.CountryRepository.storeCountries(CountryRepository.java:42) 
                       at com.mobile.myapp.UI.mvp.Presenters.SignUpPresenter.cacheCountries(SignUpPresenter.java:40) 
                       at com.mobile.myapp.UI.mvp.Presenters.SignUpPresenter$CountriesSubscriber.onNext(SignUpPresenter.java:60) 
                       at com.mobile.myapp.UI.mvp.Presenters.SignUpPresenter$CountriesSubscriber.onNext(SignUpPresenter.java:49) 
                       at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.drainNormal(ObservableObserveOn.java:200) 
                       at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.run(ObservableObserveOn.java:252) 
                       at io.reactivex.android.schedulers.HandlerScheduler$ScheduledRunnable.run(HandlerScheduler.java:109) 
                       at android.os.Handler.handleCallback(Handler.java:751)  
                       at android.os.Handler.dispatchMessage(Handler.java:95)  
                       at android.os.Looper.loop(Looper.java:154)  
                       at android.app.ActivityThread.main(ActivityThread.java:6077)  
                       at java.lang.reflect.Method.invoke(Native Method)  
                       at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:866)  
                       at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:756)  

并不重要，但如果你需要知道defaultSubscriber类的样子那就是：

DefaultSubscriber.java：

public class DefaultSubscriber<T> implements Observer<T> { 

Disposable disposable; 

@Override 
public void onSubscribe(@NonNull Disposable d) { 
    disposable = d; 
} 

@Override 
public void onNext(@NonNull T t) { 

} 

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

@Override 
public void onComplete() { 

} 

public void unsubscribe(){ 
    if(disposable!=null && !disposable.isDisposed()){ 
     disposable.dispose(); 
    } 
} 

}

Answer 1

这是一个常见的错误： just()将不会执行括号内的“代码”，因为just取值而不是计算。您需要fromCallable：

Observable.fromCallable(() -> db.countriesDao().addCountries(countriesList)) 

Answer 2

您也可以使用单一可观察

 Single.create(new SingleOnSubscribe<List<Long>>() { 
     @Override 
     public void subscribe(SingleEmitter<List<Long>> emitter) throws Exception { 
      try { 
       List<Long> ids = db.countriesDao().addCountries(countriesList); 
       emitter.onSuccess(ids); 
      } catch (Throwable t) { 
       emitter.onError(t); 
      } 
     }}) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .subscribeOn(Schedulers.io()) 
    .subscribeWith(new DisposableSingleObserver<Long>() { 
     @Override 
     public void onSuccess(List<Long> ids) { 

     } 

     @Override 
     public void onError(Throwable e) { 

     } 
}); 

Answer 3

更妙的是，你可以使用一个Completable。其描述：表示一个没有任何价值的计算，但是只表示完成或例外。

Completable.fromAction(() -> db.countriesDao().addCountries(list));