Observable.just(doSomeLongStuff()) 在我订阅 observable

Question

我对 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() ......

Answer 1

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

});

同样，我确信有更多方法可以做到这一点。我只是想解释这个问题并提供一些选择。

Answer 2

将您的 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