一、merge语法

    public Observable<Integer> getRxJavaCreateExampleData() {
        return Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                LogUtils.debug(TAG, "getRxJavaCreateExampleData---:" + Thread.currentThread().getName() + "--:" + 1);
                emitter.onNext(1);

                LogUtils.debug(TAG, "getRxJavaCreateExampleData---:" + Thread.currentThread().getName() + "--:" + 2);
                emitter.onNext(2);
                Thread.sleep(5000);
                LogUtils.debug(TAG, "getRxJavaCreateExampleData---:" + Thread.currentThread().getName() + "--:" + 3);
                emitter.onNext(3);
                emitter.onComplete();
                LogUtils.debug(TAG, "getRxJavaCreateExampleData---:" + Thread.currentThread().getName() + "--:" + 4);
                emitter.onNext(4);

            }
        });
    }

    public Observable<String> getRxJavaStringData() {
        return Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                LogUtils.debug(TAG, "getRxJavaStringData---:" + Thread.currentThread().getName() + "--:A");
                emitter.onNext("A");
                LogUtils.debug(TAG, "getRxJavaStringData---:" + Thread.currentThread().getName() + "--:B");
                emitter.onNext("B");
                LogUtils.debug(TAG, "getRxJavaStringData---:" + Thread.currentThread().getName() + "--:C");
                emitter.onNext("C");
            }
        });
    }

    public void rxJavaMergeExample() {
        Observable.merge(model.getRxJavaCreateExampleData().subscribeOn(Schedulers.io()), model.getRxJavaStringData().subscribeOn(Schedulers.io()))
                .subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .compose(RxLifeCycleUtils.<Serializable>bindUntilEvent(view, ActivityEvent.DESTROY))
                .subscribe(new Consumer<Serializable>() {
                    @Override
                    public void accept(Serializable serializable) throws Exception {
                        LogUtils.error(TAG, "rxJavaMergeExample--:" + Thread.currentThread().getName() + "-consumer-:" + serializable);
                    }
                });

    }

日志

08-18 17:11:43.814 5333-5729/com.example.zhang D/MainModel: getRxJavaCreateExampleData---:RxCachedThreadScheduler-2--:1
    getRxJavaCreateExampleData---:RxCachedThreadScheduler-2--:2
08-18 17:11:43.814 5333-5730/com.example.zhang D/MainModel: getRxJavaStringData---:RxCachedThreadScheduler-3--:A
    getRxJavaStringData---:RxCachedThreadScheduler-3--:B
    getRxJavaStringData---:RxCachedThreadScheduler-3--:C
08-18 17:11:43.819 5333-5333/com.example.zhang E/MainPresenter: rxJavaMergeExample--:main-consumer-:1
    rxJavaMergeExample--:main-consumer-:2
    rxJavaMergeExample--:main-consumer-:A
    rxJavaMergeExample--:main-consumer-:B
    rxJavaMergeExample--:main-consumer-:C
08-18 17:11:48.816 5333-5729/com.example.zhang D/MainModel: getRxJavaCreateExampleData---:RxCachedThreadScheduler-2--:3
08-18 17:11:48.817 5333-5729/com.example.zhang D/MainModel: getRxJavaCreateExampleData---:RxCachedThreadScheduler-2--:4
08-18 17:11:48.817 5333-5333/com.example.zhang E/MainPresenter: rxJavaMergeExample--:main-consumer-:3

总结

1、把多个Observable串联起来
2、它和 concat 的区别在于，不用等到 发射器 A 发送完所有的事件再进行发射器 B 的发送。

河马过河微信公众号.jpg