Android ：RxJava学习笔记之Subject

Posted 2021-12-07 JMW1407

Subject

• 1、Subject概述
• 2、种类
• 3、使用
• • 1、AsyncSubject
  • 2、BehaviorSubject
  • 3、PublishSubject
  • 4、ReplaySubject
• 参考

1、Subject概述

Subject可以看成是一个桥梁或者代理，它同时充当了Observer和Observable的角色。因为它是一个Observer，它可以订阅一个或多个Observable；又因为它是一个Observable，它可以转发它收到(Observe)的数据，也可以发射新的数据。

由于一个Subject订阅一个Observable，它可以触发这个Observable开始发射数据（如果那个Observable是"冷"的–就是说，它等待有订阅才开始发射数据）。因此有这样的效果，Subject可以把原来那个"冷"的Observable变成"热"的。

2、种类

Subject包含四种类型分别是AsyncSubject、BehaviorSubject、ReplaySubject和PublishSubject

值得注意的是一定要用Subcect.create()的方式创建并使用，不要用just(T)、from(T)、create(T)创建，否则会导致失效…

3、使用

1、AsyncSubject

简单的说使用AsyncSubject无论输入多少参数，永远只输出最后一个参数。

一个AsyncSubject只在原始Observable完成后，发射来自原始Observable的最后一个值。（如果原始Observable没有发射任何值，AsyncObject也不发射任何值）它会把这最后一个值发射给任何后续的观察者。

然而，如果原始的Observable因为发生了错误而终止，AsyncSubject将不会发射任何数据，只是简单的向前传递这个错误通知。

1、AsyncSubject当做Observable

 AsyncSubject<String> asyncSubject = AsyncSubject.create();
    asyncSubject.subscribe(
        s -> System.out.println("sucess: " + s),
        throwable -> System.out.println(throwable.getMessage()),
        () -> System.out.println("onComplete")
    );
    asyncSubject.onNext("1");
    asyncSubject.onNext("2");
    asyncSubject.onNext("3");
    asyncSubject.onComplete();

输出：

sucess: 3
onComplete

不要这样使用Subject

AsyncSubject
        .just("1", "2", "3")
        .subscribe(
        s -> System.out.println("sucess: " + s),
        throwable -> System.out.println(throwable.getMessage()),
        () -> System.out.println("onComplete")
    );

输出

sucess: 1
sucess: 2
sucess: 3
onComplete

因为just(T)、from(T)、create(T)会把Subject转换为Obserable

2、AsyncSubject当做Observer

AsyncSubject<String> asyncSubject = AsyncSubject.create();
    Observable
        .just("1", "2", "3")
        .subscribe(asyncSubject);

    asyncSubject
        .doOnComplete(() -> System.out.println("onComplete"))
        .subscribe(
            s -> System.out.println("sucess: " + s),
            throwable -> System.out.println(throwable.getMessage())
        );

输出

sucess: 3
onComplete

如果Observable因为错误发了终止：

 Observable
        .create(new ObservableOnSubscribe<String>() {
          @Override
          public void subscribe(ObservableEmitter<String> emitter) throws Exception {
            emitter.onNext("1");
            emitter.onError(new Exception("AsyncSubject"));
            emitter.onNext("2");
            emitter.onNext("3");
            emitter.onComplete();
          }
        })
        .subscribe(asyncSubject);

    asyncSubject
        .doOnComplete(() -> System.out.println("onComplete"))
        .subscribe(
            s -> System.out.println("sucess: " + s),
            throwable -> System.out.println("onError:" + throwable.getMessage())
        );

输出：

onError:AsyncSubject

2、BehaviorSubject

BehaviorSubject会发送离订阅最近的上一个值，没有上一个值的时候会发送默认值。

当观察者订阅BehaviorSubject时，它开始发射原始Observable最近发射的数据（如果此时还没有收到任何数据，它会发射一个默认值），然后继续发射其它任何来自原始Observable的数据。然而，如果原始的Observable因为发生了一个错误而终止，BehaviorSubject将不会发射任何数据，只是简单的向前传递这个错误通知。

 private void testBehaviorSubject(){
        BehaviorSubject<String> behaviorSubject = BehaviorSubject.create("default");
        Subscriber subscriber = new Subscriber<String>() {
            @Override
            public void onCompleted() {
                System.out.println("onCompleted");
            }
 
            @Override
            public void onError(Throwable e) {
                System.out.println("onError:"+e.getMessage());
            }
 
            @Override
            public void onNext(String s) {
                System.out.println("onNext:"+s);
            }
        };
 
        behaviorSubject.onNext("1");
        behaviorSubject.onNext("2");
        behaviorSubject.onNext("3");
        behaviorSubject.subscribe(subscriber);
        behaviorSubject.onNext("4");
        behaviorSubject.onNext("5");
        behaviorSubject.onCompleted();
    }

输出的结果是：

onNext:3
onNext:4
onNext:5
onCompleted

3、PublishSubject

可以说是最正常的Subject，从那里订阅就从那里开始发送数据

PublishSubject只会把在订阅发生的时间点之后来自原始Observable的数据发射给观察者。需要注意的是，PublishSubject可能会一创建完成就立刻开始发射数据（除非你可以阻止它发生），因此这里有一个风险：在Subject被创建后到有观察者订阅它之前这个时间段内，一个或多个数据可能会丢失。如果要确保来自原始Observable的所有数据都被分发，你需要这样做：或者使用Create创建那个Observable以便手动给它引入"冷"Observable的行为（当所有观察者都已经订阅时才开始发射数据），或者改用ReplaySubject。

如果原始的Observable因为发生了一个错误而终止，PublishSubject将不会发射任何数据，只是简单的向前传递这个错误通知。

PublishSubject bs = PublishSubject.create();
// 这里订阅接收1， 2， 3
bs.onNext(1);
// 这里订阅接收2， 3
bs.onNext(2);
// 这里订阅接收3
bs.onNext(3);
bs.onCompleted();
// 这里订阅无接收
bs.subscribe(
        new Action1<Integer>() {
            @Override
            public void call(Integer o) {
                LogHelper.e("S:" + o);
            }
        });

4、ReplaySubject

无论何时订阅，都会将所有历史订阅内容全部发出。

ReplaySubject会发射所有来自原始Observable的数据给观察者，无论它们是何时订阅的。也有其它版本的ReplaySubject，在重放缓存增长到一定大小的时候或过了一段时间后会丢弃旧的数据（原始Observable发射的）。

如果你把ReplaySubject当作一个观察者使用，注意不要从多个线程中调用它的onNext方法（包括其它的on系列方法），这可能导致同时（非顺序）调用，这会违反Observable协议，给Subject的结果增加了不确定性。

ReplaySubject bs = ReplaySubject.create();
// 无论何时订阅都会收到1，2，3
bs.onNext(1);
bs.onNext(2);
bs.onNext(3);
bs.onCompleted();
bs.subscribe(
        new Action1<Integer>() {
            @Override
            public void call(Integer o) {
                LogHelper.e("S:" + o);
            }
        });

总结

总的来说Subject没法指定异步线程，更像是EventBus通过订阅来实现事件通知。

参考

1、https://blog.csdn.net/baidu_34012226/article/details/52438902
2、https://xuexuan.blog.csdn.net/article/details/69568487
3、理解RxJava（四）Subject用法及原理分析