转载

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
正文到此结束
所属分类： Java 编程技术

热门推荐

相关文章

阿里云首购8折
说给你听

本文目录
随机标签
书籍教程
springboot-demo Java入门教程 bootstrap3 CSS Apache基础教程 php ionic 教程 Python mysql教程 eclipse Ubuntu VPS系统配置 AngularJS 教程 MongoDB教程 Struts2教程 Redis教程 Spring教程 Git教程 Jenkins进阶系列 openfire参考指南 Java设计模式 HBase教程 Maven教程 hibernate教程 Docker 教程 memcached教程 Quartz指南 Hive教程 Ant教程 ANTLR教程 SpringCloud java实例教程 springcloud-demo XStream教程 Hazelcast教程 Elastic-Job-Lite 深入浅出MyBatis SVN教程 ibaties教程 rabittmq教程 solr教程 Hadoop教程 WebService CXF学习 JPA教程 ActiveMQ中文指南 java-demo Java内存模型 dubbo教程 python3-demo Linux入门视频教程
近期评论
  1. 1 Spring Cloud Consul实现选举机制
  2. 2 Spring Boot集成ShedLock实现分布式定时任务
  3. 3 利用oss进行数据库和网站图片备份
  4. 4 Spring Cloud Stream实现数据流处理
  5. 5 Python3访问MySQL数据库快速入门Demo
  6. 6 Github开源项目作者可以免费申请 JetBrains 全家桶
  7. 7 Spring Cloud Vault快速入门Demo
  8. 8 Spring Cloud Gateway快速入门Demo
  9. 9 Spring Cloud Contract快速入门Demo
  10. 10 Spring Cloud Consul快速入门Demo
网站信息
  • 文章总数：82,714 篇
  • 文件总数：284,287 个
  • 标签总数：2,393 个
  • 分类总数：85 个
  • 留言数量：2,560 条
  • 在线人数：614 人
  • 运行天数：4,409天
  • 最后更新：2024年11月22日03点
Loading...

其他链接

关于本站

本站定位：个人技术类博客

本站作用：写博客、记日志、闲聊扯淡鼓捣技术。

问题交流

[HBLOG]公众号 HBLOG HBLOG