8章 RxJava高级用法（二）

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

CSDN学院课程地址

• RxJava2从入门到精通-初级篇:https://edu.csdn.net/course/detail/10036
• RxJava2从入门到精通-中级篇:https://edu.csdn.net/course/detail/10037
• RxJava2从入门到精通-进阶篇:https://edu.csdn.net/course/detail/10038
• RxJava2从入门到精通-源码分析篇:https://edu.csdn.net/course/detail/10138

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

image

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

image

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

image

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

image

只有当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谁