java - Observable.just(doSomeLongStuff()) 在订阅 observable 之前运行 doSomeLongStuff()

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

让我展示一下:

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

    }

这段代码几乎相同 - 它依次运行两次 doSomethingLong()...

我的期望: 1.我需要 doSomethingLong() 方法并行运行 2. 我要求在开始订阅这些方法之前解释为什么它们会运行。 3. 在这种情况下我应该如何写好代码。我希望在订阅 doSomethingLong() 方法之前不要调用它们。

非常感谢。希望我能很好地解释问题。

Answer 1

Observable.just 在您订阅时不会运行任何内容。当您订阅时，它会发出元素，但是您的 doSomethingLong 将在您将其作为参数传递后立即运行。这是正常的，这就是语言的工作原理。

您正在寻找的是一种在我们订阅时返回此内容的方法，但也仅在那时运行它，并希望在后台线程上运行。

对此有几个答案，以下是一些:

使用延迟

有一个名为 defer 的运算符，它接受一个 lambda，一旦您订阅，就会执行该 lambda:

Observable.defer(() ->  doSomethingLong())

这只会在您订阅时执行doSomethingLong

使用 fromCallable

您可以从 lambda 创建可观察对象。这称为 fromCallable:

Observable.fromCallable(() -> doSomethingLong())

同样，这只会在您订阅时运行 doSomethingLong

使用创建

我认为这可能是最令人沮丧的方法，因为您必须处理一些事情，但我认为为了完整性，可以提及:

Observable.create( emitter -> {
    if(emitter.isDisposed()) return;

    emitter.onNext(doSomethingLong());
    emitter.onComplete();
});

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