使用 Rx 中預(yù)定義的 Subject

看到一幅有趣的關(guān)于 Rx 學(xué)習(xí)的圖，想知道學(xué)習(xí) Rx 的學(xué)習(xí)曲線？不，是峭壁！

我們可以直接通過 Rx 的 Observer 來創(chuàng)建 Observable 對(duì)象。

但是，使用這種方式往往比較復(fù)雜，在特定的場景下，我們可以直接使用 Rx 提供的特定 Subject 來實(shí)現(xiàn) Observable。這些特定的 Subject 是主題和訂閱者的混合體，我們可以直接使用這樣的一個(gè)對(duì)象來實(shí)現(xiàn)信息的發(fā)布和數(shù)據(jù)流的訂閱。

1. Subject

通用的 Subject，既可以被訂閱，從名字也可以看到它本身就是一個(gè)主題，所以可以直接用來發(fā)布信息。如果需要實(shí)現(xiàn)一個(gè)普通的主題，它就是最理想的選擇。

使用方式：

發(fā)布信息的方法：

onNext( value )

發(fā)布一個(gè)新的值到數(shù)據(jù)流中。

onCompleted()

數(shù)據(jù)流終止。

onError( error )

發(fā)布異常。

使用示例：

var subject = new Rx.Subject();var subscription = subject.subscribe(    function (x) {
        console.log('Next: ' + x.toString());
    },    function (err) {
        console.log('Error: ' + err);
    },    function () {
        console.log('Completed');
    });

subject.next(42);// => Next: 42subject.next(56);// => Next: 56subject.completed();// => Completed

See Also: 

https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/subjects/subject.md

2. AsyncSubject

緩存直到 completed() 的最后一個(gè)值。所有的訂閱者都會(huì)收到同樣的最后一個(gè)值。

注意只能有一個(gè)值，在 completed() 之后，將不能再發(fā)布新的值。而所有的訂閱者也只能得到最后一個(gè)值。

使用方式：

必須使用 completed() 完成流，訂閱者將會(huì)在 completed() 之后才能得到最后一個(gè)值。

使用示例：

var subject = new Rx.AsyncSubject();var i = 0;var handle = setInterval(function () {
    subject.onNext(i);    if (++i > 3) {
        subject.onCompleted();
        clearInterval(handle);
    }
}, 500);var subscription = subject.subscribe(    function (x) {
        console.log('Next: ' + x.toString());
    },    function (err) {
        console.log('Error: ' + err);
    },    function () {
        console.log('Completed');
    });// => Next: 3// => Completed

圖例

See also: https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/subjects/asyncsubject.md

3. BehaviorSubject

如果是希望訂閱者獲得當(dāng)前的最后一個(gè)值，但是后面可能還會(huì)提供新的值，可以考慮這個(gè)。

緩存已經(jīng)發(fā)布的最后數(shù)據(jù)，新的訂閱者可以接收到最后一個(gè)已經(jīng)發(fā)布的值，和以后發(fā)布的新的值。

它可以直接設(shè)置一個(gè)初始值。如果不需要初始值，可以考慮使用 ReplaySubject.

使用方式：

BehaviorSubject(initialValue)

在構(gòu)造函數(shù)中提供初始的值。

getValue()

獲取當(dāng)前的值，或者拋出異常，在調(diào)用了 completed() 之后，最后的值被保留。在 error() 之后，總是拋出一個(gè)特定的異常。

使用示例

/* Initialize with initial value of 42 */var subject = new Rx.BehaviorSubject(42);var subscription = subject.subscribe(    function (x) {
        console.log('Next: ' + x.toString());
    },    function (err) {
        console.log('Error: ' + err);
    },    function () {
        console.log('Completed');
    });// => Next: 42subject.next(56);// => Next: 56subject.completed();// => Completed

圖例

See also: 

https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/subjects/behaviorsubject.md

4. ReplaySubject

可以用來緩存流中最后 n 個(gè)數(shù)據(jù)，在新的觀察者登記之后，這些緩存的數(shù)據(jù)直接發(fā)布給新的觀察者。

使用說明：

在構(gòu)造 ReplaySubject 對(duì)象的時(shí)候，配置緩存的數(shù)據(jù)元素?cái)?shù)量以及時(shí)間窗口。

ReplaySubject([bufferSize], [windowSize], [scheduler])

使用特定的緩存大小，時(shí)間窗口和調(diào)度器來創(chuàng)建 ReplaySubject 對(duì)象實(shí)例.

Arguments

1. [bufferSize = Number.MAX_VALUE] (Number): Maximum element count of the replay buffer.

2. [windowSize = NUMBER.MAX_VALUE] (Number): Maximum time length of the replay buffer.

3. [scheduler = Rx.Scheduler.currentThread] (Scheduler): Scheduler the observers are invoked on.

使用示例

var subject = new Rx.ReplaySubject(2 /* buffer size */);

subject.next('a');
subject.next('b');
subject.next('c');var subscription = subject.subscribe(    function (x) {
        console.log('Next: ' + x.toString());
    },    function (err) {
        console.log('Error: ' + err);
    },    function () {
        console.log('Completed');
    });// => Next: b// => Next: csubject.next('d');// => Next: d