上佈置

Question

在調試rxJava網絡電話在一個應用程序停止rxJava觀察到鏈執行我碰到這樣一種情況，如果我們dispose或clear處置對象由observables那麼只有第一observable鏈的訂閱返回得到處理而不是其他鏈接observables由flatMap。

看一看下面的演示代碼片段：

CompositeDisposable testCompositeDisposal = new CompositeDisposable(); 

private void testLoadData() { 
    Disposable disposable = Observable.create(sbr -> { 
     for (int i = 0; i < 5; i++) { 
      Thread.sleep(3000); 
      Log.w("Debug: ", "First: " + i); 
      sbr.onNext(true); 
     } 
     sbr.onComplete(); 
    }).subscribeOn(Schedulers.io()).flatMap(value -> Observable.create(sbr -> { 
     for (int i = 0; i < 5; i++) { 
      Thread.sleep(3000); 
      Log.w("Debug: ", "Second: " + i); 
      sbr.onNext(true); 
     } 
     sbr.onComplete(); 
    })).doOnNext(value -> { 
     Log.w("Debug: ", "doONNext"); 
    }).doOnDispose(()-> { 
     Log.w("Debug: ", "doOnDispose: observable has been disposed"); 
    }).subscribe(); 

    testCompositeDisposal.add(disposable); 
} 

@Override 
public void onStop() { 
    super.onStop(); 
    testCompositeDisposal.clear(); 
} 

輸出：

W/Debug:: First: 0 
W/Debug:: doOnDispose: observable has been disposed // I dispose Observable chain here. 
W/Debug:: First: 1 
W/Debug:: First: 2 
W/Debug:: First: 3 
W/Debug:: First: 4 

正如你可以看到上面的日誌輸出，當我處理給rxJava觀察到鏈只有第一可觀察的停止發射物品。

我想停止所有被鏈接的觀測值。

解決此問題的慣用方法是什麼？

Answer 1

兩件事情：

• flatMap可以預先消耗物品從上游（最多16上機器人）;
• 在調用onNext之前，您應該檢查觀察者是否處置（通過.isDisposed()）並在這種情況發生時中止，其次是更適用於您的用例。

此外，秒flatMap被終止（實際上它永遠不會被調用）。 第一個繼續。

編輯

private void testLoadData() { 
    Disposable disposable = Observable.create(sbr -> { 
     for (int i = 0; i < 5; i++) { 
      if(sbr.isDisposed()) return; // this will cause subscription to terminate. 
      Thread.sleep(3000); 
      Log.w("Debug: ", "First: " + i); 
      sbr.onNext(true); 
     } 
     sbr.onComplete(); 
    }).subscribeOn(Schedulers.io()).flatMap(value -> Observable.create(sbr -> { 
     for (int i = 0; i < 5; i++) { 
      Thread.sleep(3000); 
      Log.w("Debug: ", "Second: " + i); 
      sbr.onNext(true); 
     } 
     sbr.onComplete(); 
    })).doOnNext(value -> { 
     Log.w("Debug: ", "doONNext"); 
    }).doOnDispose(()-> { 
     Log.w("Debug: ", "doOnDispose: observable has been disposed"); 
    }).subscribe(); 

    testCompositeDisposal.add(disposable); 
}