转载

RxJava(八):合并操作符和连接操作符

博客主页

RxJava 的合并操作符主要包括如下几个:

  • startWith :在数据序列的开头增加一项数据
  • merge :将多个 Observable 合并为一个
  • mergeDelayError :合并多个 Observable,让没有错误的 Observable 都完成后再发射错误通知
  • zip :使用一个函数组合多个 Observable 发射的数据集合,然后再发射这个结果
  • combineLatest :当两个 Observable 中的任何一个发射了一个数据时,通过一个指定的函数组合每个 Observable 发射的最新数据( 一共两个数据),然后发射这个函数的结果
  • join and groupJoin :无论何时,如果一个 Observable 发射了一个数据项,就需要在另一个 Observable 发射的数据项定义的时间窗口内,将两个 Observable 发射的数据合并发射
  • switchOnNext :将一个发射 Observable 的 Observable 转换成另一个 Observable,后者发射这些 Observable 最近发射的数据

RxJava 的连接操作符,主要是 ConnectableObservable 所使用的操作符和 Observable 所使用

的操作符:

  • ConnectableObservable.connect : 指示一个可连接的 Observable 开始发射数据
  • Observable.publish :将一个 Observable 转换为一个可连接的 Observable
  • Observable.replay :确保所有的订阅者看到相同的数据序列,即使它们在 Observable 开始发射数据之后才订阅
  • ConnectableObservable.refCount :让一个可连接的 Observable 表现得像一个普通的 Observable

1. merge 和 zip

1.1 merge 操作符

合并多个 Observable 的发射物

RxJava(八):合并操作符和连接操作符

merge 操作符可以将多个 Observable 的输出合井,使得它们就像是单个的 Observable 一样。

Observable.merge(
        Observable.just(1, 3, 5, 7, 9),
        Observable.just(2, 4, 6)
).subscribe(new Consumer<Integer>() {
    @Override
    public void accept(Integer integer) throws Exception {
        Log.d(TAG, "Next: " + integer);
    }
}, new Consumer<Throwable>() {
    @Override
    public void accept(Throwable throwable) throws Exception {
        Log.d(TAG, "Error: " + throwable);
    }
}, new Action() {
    @Override
    public void run() throws Exception {
        Log.d(TAG, "Complete.");
    }
});

// 执行结果
 Next: 1
 Next: 3
 Next: 5
 Next: 7
 Next: 9
 Next: 2
 Next: 4
 Next: 6
 Complete.

merge 是按照时间线并行的。如果传递给 merge 的任何一个 Observable 发射了 onError 通知终止,则 merge 操作符生成的 Observable 也会立即以 onError 通知终止。如果想让它继续发射数据,直到最后才报告错误,则可以使用 mergeDelayError 操作符.

RxJava(八):合并操作符和连接操作符

merge 操作符最多只能合并 4 个被观察者,如果需要合并更多个被观察者,则可以使用 mergeArray 操作符.

1.2 zip

通过一个函数将多个 Observable 的发射物结合到一起,基于这个函数的结果为每个结合体发射单个数据项

RxJava(八):合并操作符和连接操作符

zip 操作符返回一个 Obversable ,它使用这个函数按顺序结合两个或多个 Observable 发射的数据项,然后发射这个函数返回的结果。它按照严格的顺序应用这个函数,只发射与发射数据项最少的那个 Observable 一样多的数据。

zip 的最后一个参数接收每个 Observable 发射的一项数据,返回被压缩后的数据,它可以接收 1~9 个参数:一个 Observable 序列 或者一些发射 Observable 的 Observable

Observable.zip(
        Observable.just(1, 3, 5, 7),
        Observable.just(2, 4, 6),
        new BiFunction<Integer, Integer, Integer>() {
            @Override
            public Integer apply(Integer integer, Integer integer2) throws Exception {
                return integer + integer2;
            }
        }
).subscribe(new Consumer<Integer>() {
    @Override
    public void accept(Integer integer) throws Exception {
        Log.d(TAG, "Next: " + integer);
    }
}, new Consumer<Throwable>() {
    @Override
    public void accept(Throwable throwable) throws Exception {
        Log.d(TAG, "Error: " + throwable);
    }
}, new Action() {
    @Override
    public void run() throws Exception {
        Log.d(TAG, "Complete.");
    }
});

// 执行结果
 Next: 3
 Next: 7
 Next: 11
 Complete.

zip 操作符相对于 merge 操作符,除发射数据外,还会进行合并操作,而且 zip 发射的数据与数据项最少的 Observable 有关。

这里 BiFunction 相当于一个合并函数,并不一定要返回 Integer 类型,可以根据业务需要返回合适的类型。 BiFunction 的源码如下:

public interface BiFunction<T1, T2, R> {

    /**
     * Calculate a value based on the input values.
     * @param t1 the first value
     * @param t2 the second value
     * @return the result value
     * @throws Exception on error
     */
    @NonNull
    R apply(@NonNull T1 t1, @NonNull T2 t2) throws Exception;
}

RxJava 2.x FuncN 遵循 Java 8 的命名规范。相对于 RxJava l.x, Func 改名成 Function, Func2 改名成 BiFunction, Func3~ Func9 改名成 Function3~ Function9,FuncN 由 Function 取代

2. combineLatest 和 join

2.1 combineLatest 操作符

combineLatest 操作符的行为类似于 zip,但是只有当原始的 Observable 中的每一个都发射了一条数据时 zip 才发射数据,而 combineLatest 是当原始的 Observable 中任意一个发射了数据时就发射一条数据。当原始 Observable 的任何一个发射了一条数据时, combineLatest 使用一个函数结合它们最近发射的数据,然后发射这个函数的返回值。

RxJava(八):合并操作符和连接操作符

Observable.combineLatest(
        Observable.just(1, 3, 5),
        Observable.just(2, 4, 6),
        new BiFunction<Integer, Integer, Integer>() {
            @Override
            public Integer apply(Integer integer, Integer integer2) throws Exception {
                Log.d(TAG, "integer: " + integer + " ## integer2: " + integer2);
                return integer + integer2;
            }
        }
).subscribe(new Consumer<Integer>() {
    @Override
    public void accept(Integer integer) throws Exception {
        Log.d(TAG, "Next: " + integer);
    }
}, new Consumer<Throwable>() {
    @Override
    public void accept(Throwable throwable) throws Exception {
        Log.d(TAG, "Error: " + throwable);
    }
}, new Action() {
    @Override
    public void run() throws Exception {
        Log.d(TAG, "Complete.");
    }
});


// 执行结果
 integer: 5 ## integer2: 2
 Next: 7
 integer: 5 ## integer2: 4
 Next: 9
 integer: 5 ## integer2: 6
 Next: 11
 Complete.

2.2 join 操作符

join 操作符结合两个 Observable 发射的数据,基于时间窗口(针对每条数据特定的原则)选择待集合的数据项。将这些时间窗口实现为一些 Observable ,它们的生命周期从任何一条 Observable 发射的每一条数据开始。当这个定义时间窗口的 Observable 发射了一条数据或者完成时,与这条数据关联的窗口也会关闭。只要这条数据的窗口是打开的,它就继续结合其他 Observable 发射的任何数据项。

RxJava(八):合并操作符和连接操作符

Observable<Integer> o1 = Observable.just(1, 2, 3);
Observable<Integer> o2 = Observable.just(4, 5, 6);

o1.join(o2, new Function<Integer, ObservableSource<String>>() {
    @Override
    public ObservableSource<String> apply(Integer integer) throws Exception {
        return Observable.just(String.valueOf(integer)).delay(200, TimeUnit.MILLISECONDS);
    }
}, new Function<Integer, ObservableSource<String>>() {
    @Override
    public ObservableSource<String> apply(Integer integer) throws Exception {
        return Observable.just(String.valueOf(integer)).delay(200, TimeUnit.MILLISECONDS);
    }
}, new BiFunction<Integer, Integer, String>() {
    @Override
    public String apply(Integer integer, Integer integer2) throws Exception {
        return integer + ":" + integer2;
    }
}).subscribe(new Consumer<String>() {
    @Override
    public void accept(String s) throws Exception {
        Log.d(TAG, "Next: " + s);
    }
});

// 执行结果
 Next: 1:4
 Next: 2:4
 Next: 3:4
 Next: 1:5
 Next: 2:5
 Next: 3:5
 Next: 1:6
 Next: 2:6
 Next: 3:6

join(Observable, Function, Function, BiFunction) 有四个参数:

  • Observable :源 Observable 需要组合 Observable ,这里可以称之为目标 Observable
  • Function: 接收从源 Observable 发射来的数据,井返回一个 Observable ,这个 Observable

的生命周期决定了源 Observable 发射数据的有效期。

  • Function :接收目标 Observable 发射的数据,井返回一个 Observable ,这个 Observable

的生命周期决定了目标 bservable 发射数据的有效期。

  • BiFunction :接收从源 Observable 和目 Observable 发射的数据,并将这两个数据组

合后返回。

join 操作符的效果类似于排列组合,把第一个数据源 A 作为基座窗口,它根据自己的节奏不断发射数据元素;第二个数据源 B,每发射一个数据,我们都把它和第一个数据源 A 中己经发射的数据进行一对一匹配。举例来说,如果某一时刻 B 发射了一个数据 "B",此时 A 己经发射了 a,b,c,d 共4个数据,那么合并操作就是把 “B” 依次与 a,b,c,d 配对,得到4组数据:[a, B], [b, B], [c, B], [d, B]

Observable<Integer> o1 = Observable.just(1, 2, 3).delay(200, TimeUnit.MILLISECONDS);
Observable<Integer> o2 = Observable.just(4, 5, 6);

o1.join(o2, new Function<Integer, ObservableSource<String>>() {
    @Override
    public ObservableSource<String> apply(Integer integer) throws Exception {
        return Observable.just(String.valueOf(integer)).delay(200, TimeUnit.MILLISECONDS);
    }
}, new Function<Integer, ObservableSource<String>>() {
    @Override
    public ObservableSource<String> apply(Integer integer) throws Exception {
        return Observable.just(String.valueOf(integer)).delay(200, TimeUnit.MILLISECONDS);
    }
}, new BiFunction<Integer, Integer, String>() {
    @Override
    public String apply(Integer integer, Integer integer2) throws Exception {
        return integer + ":" + integer2;
    }
}).subscribe(new Consumer<String>() {
    @Override
    public void accept(String s) throws Exception {
        Log.d(TAG, "Next: " + s);
    }
});

// 执行结果
 Next: 1:4
 Next: 1:5
 Next: 1:6
 Next: 2:4
 Next: 2:5
 Next: 2:6
 Next: 3:4
 Next: 3:5
 Next: 3:6

3. startWith

在数据序列的开头插入一条指定的项

RxJava(八):合并操作符和连接操作符

如果想让 Observable 在发射数据之前先发射一个指定的数据序列,则可以使用 startWith 操作符。如果想在一个 Observable 发射数据的末尾追加一个数据序列 ,则可以使用 concat 操作符。

Observable.just("hello, java", "hello, kotlin")
        .startWith("hello, rxjava")
        .subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d(TAG, "Next: " + s);
            }
        });

// 执行结果
 Next: hello, rxjava
 Next: hello, java
 Next: hello, kotlin

startWith 操作符支持传递 Iterable, 同时还有一个 startWithArray 的操作符

Observable.just("hello, java", "hello, kotlin")
        .startWithArray("hello, rxjava", "hello, flutter")
        .subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d(TAG, "Next: " + s);
            }
        });

// 执行结果
 Next: hello, rxjava
 Next: hello, flutter
 Next: hello, java
 Next: hello, kotlin

使用了 startWithArray 操作符之后,可以再使用 startWith 操作符。

Observable.just("hello, java", "hello, kotlin")
        .startWithArray("hello, rxjava", "hello, flutter")
        .startWith("hello, groovy")
        .subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d(TAG, "Next: " + s);
            }
        });

// 执行结果
 Next: hello, groovy
 Next: hello, rxjava
 Next: hello, flutter
 Next: hello, java
 Next: hello, kotlin

startWith 还可以传递一个 Observable ,它会将那个 Observable 的发射物插在原始 Observable 发射的数据序列之前,然后把这个当作自己的发射物集合。

Observable.just("hello, java", "hello, kotlin")
        .startWithArray("hello, rxjava", "hello, flutter")
        .startWith(Observable.just("hello, groovy"))
        .subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d(TAG, "Next: " + s);
            }
        });

// 执行结果
 Next: hello, groovy
 Next: hello, rxjava
 Next: hello, flutter
 Next: hello, java
 Next: hello, kotlin

4. connect、push 和 refCount

connect 和 refCount 是 ConnectableObservable 所使用的操作符。

ConnectableObservable 继承自 Observable ,然而它并不是在调用 subscribe() 的时候发射数据,而是只有对其使用 connect 操作符时它才会发射数据,所以可以用来更灵活地控制数据发射的时机。另外, ConnectableObservable 是 Hot Observable

push 操作符是将普通的 Observable 转换成 ConnectableObservable

connect 操作符是用来触发 ConnectableObservable 发射数据的 。我们可以等所有的观察者都订阅了 ConnectableObservable 之后再发射数据。

RxJava(八):合并操作符和连接操作符

final SimpleDateFormat format = new SimpleDateFormat("HH:mm:ss");

Observable<Long> observable = Observable.interval(1, TimeUnit.SECONDS).take(6);

ConnectableObservable<Long> connectableObservable = observable.publish();

connectableObservable.subscribe(new Observer<Long>() {
    @Override
    public void onSubscribe(Disposable d) {

    }

    @Override
    public void onNext(Long aLong) {
        Log.d(TAG, "Next#1: " + aLong + "->time:" + format.format(new Date()));
    }

    @Override
    public void onError(Throwable e) {
        Log.d(TAG, "Error: " + e);
    }

    @Override
    public void onComplete() {
        Log.d(TAG, "Complete#1.");
    }
});

connectableObservable.delaySubscription(3, TimeUnit.SECONDS)
        .subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Long aLong) {
                Log.d(TAG, "Next#2: " + aLong + "->time:" + format.format(new Date()));
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "Error: " + e);
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "Complete#2.");
            }
        });

connectableObservable.connect();

// 执行结果
 Next#1: 0->time:09:27:44
 Next#1: 1->time:09:27:45
 Next#1: 2->time:09:27:46
 Next#1: 3->time:09:27:47
 Next#2: 3->time:09:27:47
 Next#1: 4->time:09:27:48
 Next#2: 4->time:09:27:48
 Next#1: 5->time:09:27:49
 Next#2: 5->time:09:27:49
 Complete#1.
 Complete#2.

refCount 操作符是将 ConnectableObservable 转换成普通的 Observable,同时又保持了 Hot Observable 的特性。当出现第一个订阅者时, refCount 会调用 connect()。 每个订阅者每次都会接收到同样的数据,但是当所有订阅者都取消订阅(dispose)时, refCount 会自动 dispose 上游 Observable

所有的订阅者都取消订阅后,则数据流停止。如果重新订阅则数据流重新开始。如果不是所有的订阅者都取消了订阅,而是只取消了部分。则部分订阅者/观察者重新开始订阅时,不会从头开始数据流。

RxJava(八):合并操作符和连接操作符

final SimpleDateFormat format = new SimpleDateFormat("HH:mm:ss");

Observable<Long> obs = Observable.interval(1, TimeUnit.SECONDS).take(6);

ConnectableObservable<Long> connectableObservable = obs.publish();

Observable<Long> obsRefCount = connectableObservable.refCount();

obs.subscribe(new Observer<Long>() {
    @Override
    public void onSubscribe(Disposable d) {

    }

    @Override
    public void onNext(Long aLong) {
        Log.d(TAG, "Next#1: " + aLong + "->time:" + format.format(new Date()));
    }

    @Override
    public void onError(Throwable e) {
        Log.d(TAG, "Error: " + e);
    }

    @Override
    public void onComplete() {
        Log.d(TAG, "Complete#1.");
    }
});

obs.delaySubscription(3, TimeUnit.SECONDS)
        .subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Long aLong) {
                Log.d(TAG, "Next#2: " + aLong + "->time:" + format.format(new Date()));
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "Error: " + e);
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "Complete#2.");
            }
        });

obsRefCount.subscribe(new Observer<Long>() {
    @Override
    public void onSubscribe(Disposable d) {

    }

    @Override
    public void onNext(Long aLong) {
        Log.d(TAG, "Next#3: " + aLong + "->time:" + format.format(new Date()));
    }

    @Override
    public void onError(Throwable e) {
        Log.d(TAG, "Error: " + e);
    }

    @Override
    public void onComplete() {
        Log.d(TAG, "Complete#3.");
    }
});

obsRefCount.delaySubscription(3, TimeUnit.SECONDS)
        .subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Long aLong) {
                Log.d(TAG, "Next#4: " + aLong + "->time:" + format.format(new Date()));
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "Error: " + e);
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "Complete#4.");
            }
        });


// 执行结果
 Next#1: 0->time:09:39:50
 Next#3: 0->time:09:39:50
 Next#1: 1->time:09:39:51
 Next#3: 1->time:09:39:51
 Next#1: 2->time:09:39:52
 Next#3: 2->time:09:39:52
 Next#1: 3->time:09:39:53
 Next#3: 3->time:09:39:53
 Next#4: 3->time:09:39:53
 Next#2: 0->time:09:39:53
 Next#1: 4->time:09:39:54
 Next#3: 4->time:09:39:54
 Next#4: 4->time:09:39:54
 Next#2: 1->time:09:39:54
 Next#3: 5->time:09:39:55
 Next#4: 5->time:09:39:55
 Complete#3.
 Next#1: 5->time:09:39:55
 Complete#4.
 Complete#1.
 Next#2: 2->time:09:39:55
 Next#2: 3->time:09:39:56
 Next#2: 4->time:09:39:57
 Next#2: 5->time:09:39:58
 Complete#2.

5. replay

保证所有的观察者收到相同的数据序列,即使它们 Observable 开始发射数据之后才订阅

RxJava(八):合并操作符和连接操作符

replay 操作符返回一个 ConnectableObservable 对象,并且可以缓存发射过的数据,这样即使有订阅者在发射数据之后进行订阅,也能收到之前发射过的数据。不过使用 replay 操作符最好还是先限定缓存的大小,否则缓存的数据太多时会占用很大一块内存。对缓存的控制可以从空间和时间两个方面来实现。

replay 操作符生成的 ConnectableObservable ,使得观察者无论什么时候开始订阅,都能收到 Observable 发送的所有数据

final SimpleDateFormat format = new SimpleDateFormat("HH:mm:ss");

Observable<Long> obs = Observable.interval(1, TimeUnit.SECONDS).take(6);

ConnectableObservable<Long> connectableObservable = obs.replay();

connectableObservable.connect();

connectableObservable.subscribe(new Observer<Long>() {
    @Override
    public void onSubscribe(Disposable d) {

    }

    @Override
    public void onNext(Long aLong) {
        Log.d(TAG, "Next#1: " + aLong + "->time:" + format.format(new Date()));
    }

    @Override
    public void onError(Throwable e) {
        Log.d(TAG, "Error: " + e);
    }

    @Override
    public void onComplete() {
        Log.d(TAG, "Complete#1.");
    }
});

connectableObservable.delaySubscription(3, TimeUnit.SECONDS)
        .subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Long aLong) {
                Log.d(TAG, "Next#2: " + aLong + "->time:" + format.format(new Date()));
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "Error: " + e);
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "Complete#2.");
            }
        });


// 执行结果
 Next#1: 0->time:09:59:01
 Next#1: 1->time:09:59:02
 Next#1: 2->time:09:59:03
 Next#2: 0->time:09:59:03
 Next#2: 1->time:09:59:03
 Next#2: 2->time:09:59:03
 Next#1: 3->time:09:59:04
 Next#2: 3->time:09:59:04
 Next#1: 4->time:09:59:05
 Next#2: 4->time:09:59:05
 Next#1: 5->time:09:59:06
 Next#2: 5->time:09:59:06
 Complete#1.
 Complete#2.

connect() 无须在观察者订阅之后调用也能执行.

replay 有多个接收不同参数的重载方法,有的可以指定 replay 的最大缓存数量,有的可以指定调度器。

ConnectableObservable 的线程切换只能通过 replay 操作符实现,普通 Observable 的subscribeOn() 和 observerOn() 在 ConnectableObservable 中不起作用。 replay 操作符可以通过指定线程的方式来切换线程。

如果我的文章对您有帮助,不妨点个赞鼓励一下(^_^)

原文  https://segmentfault.com/a/1190000021605506
正文到此结束
Loading...