RXJava Subject详解 (RXJava Part 2)

前面说过了RXJava中的Observable，本次来说说Subject。
附上前文的链接。喜欢的兄弟麻烦点个喜欢，关注啥的吧。
RXJava Observable详解 (RXJava Part 1)

Subject

Subject有两种用途：

1. 做为observable向其他的observable发送事件
2. 做为observer接收其他的observable发送的事件。
   最后的一个例子会使用一个subject监听一个observable，并将observable发射的事件转发给一个observer。

Subject做为Observable

• PublishSubject

该Subject不会改变事件的发送顺序。
如果在已经发送了一部分事件之后注册的observer，
是不会收到之前发送的事件。

重点跟下面三个类作对比

    private void doPublishSubject() {
        //将事件发送到observer，如果先前已经漏掉的事件，不会重新发送到后注册的observer上
        PublishSubject<String> publish = PublishSubject.create();
        publish.subscribe(new PublishObserver<String>("first"));
        publish.onNext("1");
        publish.onNext("2");
        publish.subscribe(new PublishObserver<String>("seconde"));
        publish.onNext("3");
        publish.onCompleted();
    }
    
    class SubjectObserver<T extends String> implements Observer<String> {

    String name;

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

    @Override
    public void onCompleted() {
        LogUtils.d("publishObserver %s is completed", name);
    }

    @Override
    public void onError(Throwable e) {
        LogUtils.d("publishObserver %s error msg %s", name, e.getStackTrace());
    }

    @Override
    public void onNext(java.lang.String s) {
        LogUtils.d("publishObserver %s receive msg %s", name, s);
    }
}

• BehaviorSubject

该类有创建时需要一个默认参数，该默认参数会在subject未发送过其他的事件时，向注册的observer发送。
注意看代码注释。

    private void doBehaviorSubject() {
        //将事件发送到observer，如果先前已经漏掉的事件，除了最近的一个事件以外，
        //其他相关事件不会重新发送到后注册的observer上。所以需要带默认值，
        //第一次被observer注册时，observable中没有内容的时候，就会将默认值发给observer
        BehaviorSubject<String> behavior = BehaviorSubject.create("创建beahavior时候带的消息");
        behavior.subscribe(new SubjectObserver<String>("first"));
        behavior.onNext("1");
        behavior.onNext("2");
        behavior.subscribe(new SubjectObserver<String>("seconde"));
        behavior.onNext("3");
        behavior.onCompleted();
    }

• ReplaySubject

将事件发送到observer，无论什么时候注册observer，
无论何时通过该observable发射的所有事件，均会发送给新的observer。

    private void doReplaySubject() {
        //将事件发送到observer，无论什么时候注册observer，
        //无论何时通过该observable发射的所有事件，均会发送给新的observer。
        ReplaySubject<String> replay = ReplaySubject.create();
        replay.subscribe(new SubjectObserver<String>("first"));
        replay.onNext("1");
        replay.onNext("2");
        replay.subscribe(new SubjectObserver<String>("seconde"));
        replay.onNext("3");
        replay.onCompleted();
    }

• AsyncSubject

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

    private void doAsyncSubject() {
        //只会有当subject调用onComplete方法时，才会将subject中的最后一个事件传递给observer。
        //如果不调用onComplete方法，则不会向observer中发送任何事件
        AsyncSubject async = AsyncSubject.create();
        async.subscribe(new SubjectObserver<String>("first"));
        async.onNext("1");
        async.onNext("2");
        async.onNext("3");
        async.onCompleted();
        async.subscribe(new SubjectObserver<String>("seconde"));
        async.onCompleted();
    }

Subject做为observable

文章开头说了，subject及可以做observer也可以做observable。
示例会将subject注册为一个observer来接收创建的observable中的事件。
并将observable中的事件再发送给在该Subject中注册的observer中。

  private void doObserverSubject() {
        //将Subject当作Observer使用，并将另外的observer注册到subject上，来监听原始的observable发出的事件
        List<String> items = new ArrayList<>();
        items.add("100");
        items.add("103");
        items.add("107");
        Observable<String> observable = Observable.from(items);
        ReplaySubject<String> replay = ReplaySubject.create();
        observable.subscribe(replay);
        replay.subscribe(new SubjectObserver<String>("first"));
    }

---

Android交流qq群号 29552133