PublishSubject

Publish是Subject的一個基礎子類拇颅。發(fā)送訂閱后的數(shù)據(jù)流。

PublishSubject<Integer> publishSubject = PublishSubject.create();

BehaviorSubject

BehaviorSubject會首先向他的訂閱者發(fā)送截至訂閱前最新的一個數(shù)據(jù)對象（或初始值）,然后正常發(fā)送訂閱后的數(shù)據(jù)流再登。

BehaviorSubject<Integer> behaviorSubject = BehaviorSubject.create(0);

我們創(chuàng)建了一個能發(fā)射整形(Integer)的BehaviorSubject。由于每當Observes訂閱它時就會發(fā)射最新的數(shù)據(jù)，所以它需要一個初始值字旭。

ReplaySubject

ReplaySubject會緩存它所訂閱的所有數(shù)據(jù),向任意一個訂閱它的觀察者重發(fā):

ReplaySubject<Integer> replaySubject = ReplaySubject.create();

AsyncSubject

當Observable完成時AsyncSubject只會發(fā)布最后一個數(shù)據(jù)給已經(jīng)訂閱的每一個觀察者。

AsyncSubject<Integer> asyncSubject = AsyncSubject.create();

測試

    @Test
    public void observerTest() {
        PublishSubject<Integer> publishSubject = PublishSubject.create();
        BehaviorSubject<Integer> behaviorSubject = BehaviorSubject.create(0);
        ReplaySubject<Integer> replaySubject = ReplaySubject.create();
        AsyncSubject<Integer> asyncSubject = AsyncSubject.create();

        List<Subject> subjects = new ArrayList<>();
        subjects.add(publishSubject);
        subjects.add(behaviorSubject);
        subjects.add(replaySubject);
        subjects.add(asyncSubject);

        for (Subject subject :
                subjects) {
            System.out.println("--------------------------------\n" + subject.getClass().getSimpleName() + " start");
            subject.onNext(-3);
            subject.onNext(-2);
            subject.onNext(-1);
            subject.subscribe(new Subscriber<Integer>() {
                @Override
                public void onCompleted() {
                    System.out.println("onCompleted");
                }

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

                @Override
                public void onNext(Integer integer) {
                    System.out.println(integer);
                }
            });

            subject.onNext(1);
            subject.onNext(2);
            subject.onNext(3);
            subject.onCompleted();

            System.out.println(subject.getClass().getSimpleName() + " end\n================================\n");
        }
    }

--------------------------------
PublishSubject start
1
2
3
onCompleted
PublishSubject end
================================

--------------------------------
BehaviorSubject start
-1  # 如果前面沒有發(fā)送數(shù)據(jù)崖叫， 此處會顯示 0
1
2
3
onCompleted
BehaviorSubject end
================================

--------------------------------
ReplaySubject start
-3
-2
-1
1
2
3
onCompleted
ReplaySubject end
================================

--------------------------------
AsyncSubject start
3
onCompleted
AsyncSubject end
================================