当前位置：首页 > news >正文

RxJava中的Subject

news 2025/8/28 12:36:41

要使用Rxjava首先要导入两个包，其中rxandroid是rxjava在android中的扩展

implementation 'io.reactivex:rxandroid:1.2.1'implementation 'io.reactivex:rxjava:1.2.0'

Subject
Subject 既可以是一个 Observer 也可以是一个 Observerable，它是连接 Observer 和Observerable的桥梁

PublishSubject
PublishSubject只会把在订阅发生的时间点之后来自原始Observable的数据发射给观察者。需要注意的是，PublishSubject可能会一创建完成就立刻开始发射数据，因此这里会有一个风险：在Subject被创建后到有观察者订阅它之前这个时间段内，一个或多个数据可能会丢失。如果要确保来自原始Observable的所有数据都被分发，则可以当所有观察者都已经订阅时才开始发射数据，或者改用ReplaySubject。

        PublishSubject<String> publishSubject = PublishSubject.create();publishSubject.onNext("哈哈哈"); //数据不会发射给观察者publishSubject.subscribe(observer);publishSubject.onNext("哈哈哈2"); //数据会发射给观察者

BehaviorSubject
当Observer订阅BehaviorSubject时，它开始发射原始Observable最近发射的数据。如果此时还没有收到任何数据，它会发射一个默认值，然后继续发射其他任何来自原始Observable的数据。如果原始的Observable因为发生了一个错误而终止，BehaviorSubject将不会发射任何数据，但是会向Observer传递一个异常通知。

        BehaviorSubject<String> behaviorSubject = BehaviorSubject.create();behaviorSubject.onNext("哈哈0");  //数据不会发射给观察者behaviorSubject.onNext("哈哈1");  //数据会发射给观察者behaviorSubject.subscribe(observer);behaviorSubject.onNext("哈哈2");behaviorSubject.onNext("哈哈3");

如果继续引用 observable.subscribe(behaviorSubject); 则会打印

 observer onNext is come in s= 哈哈1observer onNext is come in s= 哈哈2observer onNext is come in s= 哈哈3observer onNext is come in s= observable call onNext0observer onNext is come in s= observable call onNextobserver onCompleted is come in

ReplaySubject
不管Observer何时订阅ReplaySubject，ReplaySubject均会发射所有来自原始Observable的数据给Observer。有不同类型的ReplaySubject，它们用于限定Replay的范围，例如设定Buffer的具体大小，或者设定具体的时间范围。如果使用ReplaySubject作为Observer，注意不要在多个线程中调用onNext、onComplete和onError方法。这可能会导致顺序错乱，并且违反了Observer规则。

        ReplaySubject<String> replaySubject = ReplaySubject.create();replaySubject.onNext("哈哈0");  //数据会发送给观察者replaySubject.onNext("哈哈1"); //数据会发送给观察者replaySubject.subscribe(observer);replaySubject.onNext("哈哈2");replaySubject.onNext("哈哈3");

AsyncSubject
当Observable完成时(必须调用onCompleted()才会发射数据)，AsyncSubject只会发射来自原始Observable的最后一个数据(不管数据在订阅前还是订阅后)。如果原始的 Observable 因为发生了错误而终止，AsyncSubject 将不会发射任何数据，但是会向Observer传递一个异常通知。

        AsyncSubject<String> asyncSubject = AsyncSubject.create();asyncSubject.onNext("哈哈0");//数据不会发送给观察者asyncSubject.onNext("哈哈1");//数据不会发送给观察者asyncSubject.onNext("哈哈2");//数据不会发送给观察者(如果没有哈哈3和哈哈4 则本条数据会发射给观察者)asyncSubject.subscribe(observer);asyncSubject.onNext("哈哈3"); //数据不会发送给观察者asyncSubject.onNext("哈哈4"); //数据会发送给观察者asyncSubject.onCompleted(); //数据会发送给观察者