RxJava 合并订阅者仅从第一个 observable 中获取结果

Question

我试图了解 rxjava 合并是如何工作的。所以这里是简单的代码，它应该合并 2 个 observables 的结果并发送给订阅者

    Observable.merge(getObservable(), getTimedObservable())
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<String>() {
                    @Override public void call(final String s) {
                        Log.i("test", s);
                    }
                });

    private Observable<String> getTimedObservable() {
        return Observable.interval(150, TimeUnit.MILLISECONDS)
                .map(new Func1<Long, String>() {
                    @Override public String call(final Long aLong) {
                        Log.i("test", "tick thread: " + Thread.currentThread().getId());
                        return String.valueOf(aLong);
                    }
                });
    }

    public Observable<String> getObservable() {
        return  Observable.create(new Observable.OnSubscribe<String>() {
            @Override public void call(final Subscriber<? super String> subscriber) {
                try {
                    Log.i("test", "simple observable thread: " + Thread.currentThread().getId());
                    for (int i = 1; i <= 10; i++) {
                        subscriber.onNext(String.valueOf(i * 100));
                        Thread.sleep(300);
                    }
                    subscriber.onCompleted();
                } catch (Exception e) {
                    subscriber.onError(e);
                }
            }
        });
    }

我预计订阅者的合并结果会像

100 0 1 200 2 300 4 5 400

或类似的东西，但是，实际结果是：

 test: simple observable thread: 257 
test: 100
test: 200
test: 300
test: 400
test: 500
test: 600
test: 700
test: 800
test: 900
test: 1000
test: tick thread: 254
test: 0
test: tick thread: 254
test: 1
test: tick thread: 254
test: 2
test: tick thread: 254
test: 3
test: tick thread: 254
test: 4
test: tick thread: 254
test: 5
test: tick thread: 254
test: 6
test: tick thread: 254
test: 7
test: tick thread: 254
test: 8
test: tick thread: 254
test: 9
test: tick thread: 254
test: 10
test: tick thread: 254
test: 11
test: tick thread: 254
test: 12
test: tick thread: 254
test: 13

它看起来像第一个 Observable 块中的 Thread.sleep 在第二个 observable 中发射，但我不明白如何。有人可以解释一下吗？

Answer 1

merge 将同时订阅两个 observable。将首先订阅的 observable 将在调用线程时产生值。因为调用线程被 observable1 阻塞，所以 observable2 不能产生值。SubscribeOn 只会说明订阅发生的位置。假设 observable 开始在 main-1 上产生值。下游的每个值都在同一个线程上。没有并发。

如果你想实现并发，你必须为每个 observable 指定订阅必须发生的地方。所以假设我们有两个 Observables 的 Observables.merge。Observable1 和 Observable2 有 subscribeOn 和一些线程池。每个 observable 将在给定的 subscribeOn 线程上生成值。你实现了并发。

请查看编辑后的输出：

@Test
public void name() throws Exception {
    Subscription subscribe = Observable.merge(getObservable(), getTimedObservable())
            //.observeOn(AndroidSchedulers.mainThread())
            .subscribe(s -> {

                System.out.println("subscription " + s);
                //Log.i("test", s);
            });


    Thread.sleep(5_000);
}

private Observable<String> getTimedObservable() {
    return Observable.interval(150, TimeUnit.MILLISECONDS)
            .map(aLong -> {
                System.out.println("getTimedObservable: " + Thread.currentThread().getId());

                //Log.i("test", "tick thread: " + Thread.currentThread().getId());
                return String.valueOf(aLong);
            }).subscribeOn(Schedulers.io());
}

private Observable<String> getObservable() {
    return Observable.<String>create(subscriber -> {
        try {
            for (int i = 1; i <= 10; i++) {
                System.out.println("getObservable: " + Thread.currentThread().getId());
                subscriber.onNext(String.valueOf(i * 100));
                Thread.sleep(300);
            }
            subscriber.onCompleted();
        } catch (Exception e) {
            subscriber.onError(e);
        }
    }).subscribeOn(Schedulers.io());
}