如何組織異步調用的鏈RxJava2方式

Question

我裹着的AsyncTask類

 AsyncTask<Void, Void, Void> task = new AsyncTask<Void, Void, Void>() { 
     @Override 
     protected Void doInBackground(Void... params) { 
      try { 
       if (NetworkUtils.isNetworkConnected(mContext)) 
        mIndicatorTable.sync().blockingGet(); 
      } catch (InterruptedException | ExecutionException e) { 
       e.printStackTrace(); 
      } finally { 
       try { 
        final MobileServiceList<IndicatorModel> results = mIndicatorTable.table().read(null).blockingGet(); 
        if (isViewAttached()) { 
         getMvpView().refreshIndicatorList(results); 
         getMvpView().hideLoading(); 
        } 
       } catch (Exception e) { 
        e.printStackTrace(); 
       } 
      } 
      return null; 
     }; 

mIndicatorTable.sync()和mIndicatorTable.table().read(null)都返回Single<Boolean>下一個代碼。 mIndicatorTable.sync() - 如果網絡不可用，則將本地存儲與遠程存儲同步，如果網絡不可用，那麼我們在網絡準備就緒時使用mIndicatorTable.table().read(null)從本地存儲讀取 - 我們執行同步，然後從本地存儲讀取（我們不關心同步取消或中斷）。畢竟，我們應該調用View來刷新RecycleView。它如何與RxJava2一起實現？

#UPDATE

工作版本

getDataManager().syncAll() 
    .onErrorResumeNext(Single.just(false)) 
    .flatMap(list -> toSingle(mIndicatorTable.getTable().read(null))) 
    .subscribeOn(Schedulers.newThread()) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .subscribe(results -> { 
     if (isViewAttached()) { 
      getMvpView().refreshIndicatorList(results); 
      getMvpView().hideLoading(); 
     } 
    }) 

Answer 1

RX-java的化妝鏈的異步調用的只是打電話.flatMap()容易。

mIndicatorTable.sync() 
     .flatMap(new Function<Boolean, SingleSource<MobileServiceList<IndicatorModel>>>() { 
      @Override 
      public SingleSource<MobileServiceList<IndicatorModel>> apply(Boolean aBoolean) throws Exception { 
       return mIndicatorTable.table().read(null); 
      } 
     }) 
     .subscribeOn(Schedulers.io()) 
     .observeOn(AndroidSchedulers.mainThread()) 
     .subscribe(...) 

但你要繼續在另一個線程中調用sync()和RecycleView只能從主線程訪問。這就是rx-android派上用場的地方。 .subscribeOn(Schedulers.io())在另一個線程中實例化流，並確保在主線程中執行此行下面的語句（這裏是.subscribe(...)）。