我對 RxJava2 有愚蠢的問題。我需要同時運行兩個長時間的操作。我知道我應該使用 Observable.zip() 并且我使用了它。問題是我的長期操作一個接一個地運行，另一個問題是我的長期操作在我訂閱它們之前就開始了。假設這是我應該異步運行的長期操作。private String doSomethingLong() { Random rand = new Random(); int value = rand.nextInt(5); Timber.i("Do something for [%d] sec [%s]", value, Thread.currentThread().getName()); try { Thread.sleep(value * 1000); } catch (InterruptedException e) { e.printStackTrace(); return String.format(Locale.getDefault(), "Exception [%s]", e.getMessage()); } return String.format(Locale.getDefault(),"Job for [%d] seconds", value); }并讓有一個像 test() 這樣的方法來嘗試使其并行：public void test() { final long started = System.currentTimeMillis(); Observable<String> just1 = Observable.just(doSomethingLong()).subscribeOn(Schedulers.newThread()); Observable<String> just2 = Observable.just(doSomethingLong()).subscribeOn(Schedulers.newThread()); Observable.zip(just1, just2, new Func2<String, String, Combined>() { @Override public Combined call(String s, String s2) { return new Combined(s, s2); } }).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<Combined>() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(Combined combined) { long total = System.currentTimeMillis() - started; Timber.i("TOTAL [%d]ms [%s]", total, combined.toString()); } }); }當我嘗試運行它時，我觀察到兩個可觀察對象 just1 和 just2 一個接一個地運行……這讓我很困惑……但是還有另一個工作人員讓我更加困惑......我評論了 Observable.zip 并注意到 just1 和 just2 在我訂閱它們之前啟動了方法 doSomethingLong() ......
2 回答
汪汪一只貓
TA貢獻1898條經驗 獲得超8個贊
Observable.just訂閱時不運行任何內容。它會在您訂閱時發出元素，但您doSomethingLong將在將其作為參數傳遞后立即運行。這很正常，這就是語言的工作方式。
你正在尋找的是一種在我們訂閱時返回它的方法，但也只在那個時候運行它，并希望在后臺線程上運行。
有幾個答案，這里有一些：
使用延遲
有一個名為的運算符defer，它接受一個 lambda，它會在您訂閱后執行：
Observable.defer(() -> doSomethingLong())
這只會doSomethingLong在您訂閱時執行
使用 fromCallable
您可以從 lambda 創建一個可觀察對象。這被稱為fromCallable：
Observable.fromCallable(() -> doSomethingLong())
同樣，這只會doSomethingLong在您訂閱時運行
使用創建
我認為這可能是最令人氣餒的做法，因為有幾件事你必須處理，但我認為對于完整性來說，可以提一下：
Observable.create( emitter -> {
if(emitter.isDisposed()) return;
emitter.onNext(doSomethingLong());
emitter.onComplete();
});
同樣，我確信有更多方法可以做到這一點。我只是想解釋這個問題并提供一些選擇。
PIPIONE
TA貢獻1829條經驗 獲得超9個贊
將您的 Observables 創建為 Observable.fromCallable{}。而不是 zip 使用 combineLatest()
文檔： http ://reactivex.io/RxJava/javadoc/io/reactivex/Observable.html#fromCallable-java.util.concurrent.Callable- http://reactivex.io/documentation/operators/combinelatest.html
添加回答
舉報
0/150
提交
取消