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