8章 RxJava高级用法（二）

本篇文章已受权微信公众号 YYGeeker 独家发布转载请标明出处

CSDN学院课程地址

• RxJava2从入门到精通-初级篇:edu.csdn.net/course/deta…
• RxJava2从入门到精通-中级篇:edu.csdn.net/course/deta…
• RxJava2从入门到精通-进阶篇:edu.csdn.net/course/deta…
• RxJava2从入门到精通-源码分析篇:edu.csdn.net/course/deta…

8. RxJava高级用法（二）

8.1 Subject的介绍

Subject能够当作是一个桥梁或者代理，在RxJava中同时充当了Observer和Observable的角色，即事件流的发送方和接收方。由于它是一个Observer，它能够订阅一个或多个Observable；又由于它是一个Observable，它能够转发它收到(Observe)的数据，也能够发射新的数据。为了方便演示，在开始讲解Subject以前，咱们统一建立出SubjectObserver

static class SubjectObserver<T extends String> implements Observer<String> {

    String name;

    public SubjectObserver(String name) {
        this.name = name;
    }

    @Override
    public void onError(Throwable e) {
        System.out.println("onError " + e.getStackTrace());
    }

    @Override
    public void onComplete() {
        System.out.println(name + " onComplete");
    }

    @Override
    public void onSubscribe(Disposable d) {

    }

    @Override
    public void onNext(java.lang.String s) {
        System.out.println(name + " onNext=" + s);
    }
}
复制代码

8.2 PublishSubject

PublishSubject不会改变事件的发送顺序。若是在已经发送了一部分事件以后注册的observer，是不会收到以前发送的事件

public static void publishSubject() {
    PublishSubject publishSubject = PublishSubject.create();
    publishSubject.subscribe(new SubjectObserver<>("first"));
    publishSubject.onNext("1");
    publishSubject.onNext("2");
    publishSubject.subscribe(new SubjectObserver<>("seconde"));
    publishSubject.onNext("3");
    publishSubject.onComplete();
}
复制代码

输出结果

first onNext=1
first onNext=2
first onNext=3
seconde onNext=3
first onComplete
seconde onComplete
复制代码

8.3 BehaviorSubject

BehaviorSubject会建立出带有默认值的事件流。当BehaviorSubject第一次被observer注册时，若是observable中没有发射数据项的时候，就会将默认值发给observer；若是observable中有发射过数据项的时候，就会将最近发射的数据项发给observer

public static void behaviorSubject(View view) {
    BehaviorSubject<String> behaviorSubject = BehaviorSubject.createDefault("默认值");
    behaviorSubject.subscribe(new SubjectObserver<String>("first"));
    behaviorSubject.onNext("1");
    behaviorSubject.onNext("2");
    behaviorSubject.subscribe(new SubjectObserver<String>("seconde"));
    behaviorSubject.onNext("3");
    behaviorSubject.onComplete();
}
复制代码

输出结果

first onNext=默认值
first onNext=1
first onNext=2
seconde onNext=2
first onNext=3
seconde onNext=3
first onComplete
seconde onComplete
复制代码

8.4 ReplaySubject

ReplaySubject不管何时注册observer，不管什么时候经过ReplaySubject发射的全部事件，均会发送给新的observer

public static void replaySubject(View view) {
    ReplaySubject<String> replaySubject = ReplaySubject.create();
    replaySubject.subscribe(new SubjectObserver<>("first"));
    replaySubject.onNext("1");
    replaySubject.onNext("2");
    replaySubject.subscribe(new SubjectObserver<>("seconde"));
    replaySubject.onNext("3");
    replaySubject.onComplete();
}
复制代码

输出结果

first onNext=1
first onNext=2
seconde onNext=1
seconde onNext=2
first onNext=3
seconde onNext=3
first onComplete
seconde onComplete
复制代码

8.5 AsyncSubject

只有当AsyncSubject调用onComplete方法时，才会将AsyncSubject中的最后一个事件传递给observer。若是不调用onComplete方法，则不会给observer发送任何事件

public static void asyncSubject(View view) {
    AsyncSubject asyncSubject = AsyncSubject.create();
    asyncSubject.subscribe(new SubjectObserver<String>("first"));
    asyncSubject.onNext("1");
    asyncSubject.onNext("2");
    asyncSubject.onNext("3");
    asyncSubject.onComplete();
    asyncSubject.subscribe(new SubjectObserver<String>("seconde"));
    asyncSubject.onComplete();
}
复制代码

输出结果

first onNext=3
first onComplete
seconde onNext=3
seconde onComplete
复制代码

8.6 转发数据

每种Subject均可以看成是转发数据的桥梁，在转发数据的过程当中，具体使用哪一种应该参考对应的应用场景，一般在组件之间的通信就须要用到Subject。好比下面这个例子，从组件A中将信息传递给组件B，组件B会收到组件A发送过来的消息，若是须要组件B将信息传递给组件B，则须要将代码的观察者角色和被观察者角色调换过来。一般状况下，组件A和组件B都会同时依附于同一个Activity或Fragment，在Activity或Fragment中应该提供获取Subject的方法，让组件双方能够获取桥梁，从而互相subscribe，最终造成通信

public static void transpondData(View view){
    //组件A
    Observable<String> observable = Observable.fromArray("123","456","789");
    ReplaySubject<String> replaySubject = ReplaySubject.create();

    observable.subscribe(replaySubject);
    replaySubject.subscribe(new SubjectObserver<>("B"));//组件B
}
复制代码

输出结果

B onNext=123
B onNext=456
B onNext=789
B onComplete
复制代码

8.7 小结

1. PublishSubject：从注册时开始接收新的数据项
2. BehaviorSubject：从注册时会收到之前最近的一条数据项，若是没有则收到默认值
3. ReplaySubject：不管什么时候注册，都会收到之前发送过的全部数据项到新的observer
4. AsyncSubject：只有调用onComplete时，才会收到最后一个数据项
5. 转发数据：A能够转发给B，B能够转发给A，主要是看谁subscribe谁