javascript - 我如何確保對觀察者的 subscribe 調用最初接收到最新的值？

我正在將 rxjs 與 Angular 2 和 Typescript 一起使用。我想在多個組件之間共享一個公共的網絡資源（在我的應用上下文中稱為“專案”，本質上是一個 JSON 文檔）。為了實現這一點，我引入了一個服務，該服務暴露了一個 observable，將被所有客戶端共享：

/**
 * Handed out to clients so they can subscribe to something.
 */
private _observable : Observable<Project>;

/**
 * Used to emit events to clients.
 */
private _observer : Observer<Project>;

constructor(private _http: Http) {
    // Create observable and observer once and for all. These instances
    // are not allowed to changed as they are passed on to every subscriber.
    this._observable = Observable.create( (obs : Observer<Project>) => {
        this._observer = obs;
    });
}

客戶現在只需獲取對那個 _observable 的引用並訂閱它。

/**
 * Retrieves an observable that always points to the active
 * project.
 */
get ActiveProject() : Observable<Project> {
    return (this._observable);
}

當某個組件決定實際載入一個專案時，它會呼叫以下方法：

/**
 * @param id The id of the project to set for all subscribers
 */
setActiveProject(id : string) {
    // Projects shouldn't change while other requests are in progress
    if (this._httpRequest) {
        throw { "err" : "HTTP request in progress" };
    }

    this._httpRequest = this._http.get('/api/project/' + id)
        .catch(this.handleError)
        .map(res => new Project(res.json()));

    this._httpRequest.subscribe(res => {
        // Cache the project
        this._cachedProject = res;
        // Show that there are no more requests
        this._httpRequest = null;
        // Inform subscribers
        this._observer.next(this._cachedProject)

        console.log("Got project");
    });
}

它基本上會執行一個 HTTP 請求，將 JSON 文件轉換成「正確的」實例，並呼叫 this._observer.next() 以通知所有訂閱者有關變更。

但是如果某些訂閱發生在 HTTP 請求已經完成之後，那麼直到發出新的 HTTP 請求之前，它們什麼都看不到。我發現 rxjs 中似乎有某種緩存（或重放？）機制可以解決這個問題，但我還沒弄明白如何使用它。

tl;dr：我如何確保對觀察者調用 subscribe 時，最初接收到最新的值？

額外問題：透過在建構函數中「將觀察者從可觀察對象中拉出來」，我是否本質上創建了一個主題？

答案＃1

這就是 BehaviorSubject 的作用

import { BehaviorSubject } from 'rxjs/subject/BehaviorSubject';
...
obs=new BehaviourSubject(4);
obs.subscribe(); //prints 4
obs.next(3); //prints 3
obs.subscribe(); //prints 3

答案＃2

我通常使用 shareReplay(1) 來達成這個目的。使用這個操作符並將參數設為 1，將確保最新發出的值被保存在緩衝區中，因此當有新的訂閱者時，該值會立即傳給它。你可以查看文檔：

var interval = Rx.Observable.interval(1000);

var source = interval
    .take(4)
    .doAction(function (x) {
        console.log('Side effect');
    });

var published = source
    .shareReplay(3);

published.subscribe(createObserver('SourceA'));
published.subscribe(createObserver('SourceB'));

// Creating a third subscription after the previous two subscriptions have
// completed. Notice that no side effects result from this subscription,
// because the notifications are cached and replayed.
Rx.Observable
    .return(true)
    .delay(6000)
    .flatMap(published)
    .subscribe(createObserver('SourceC'));

function createObserver(tag) {
    return Rx.Observer.create(
        function (x) {
            console.log('Next: ' + tag + x);
        },
        function (err) {
            console.log('Error: ' + err);
        },
        function () {
            console.log('Completed');
        });
}

// => Side effect
// => Next: SourceA0
// => Next: SourceB0
// => Side effect
// => Next: SourceA1
// => Next: SourceB1
// => Side effect
// => Next: SourceA2
// => Next: SourceB2
// => Side effect
// => Next: SourceA3
// => Next: SourceB3
// => Completed
// => Completed
// => Next: SourceC1
// => Next: SourceC2
// => Next: SourceC3
// => Completed

額外問題：通過「將觀察者從可觀察對象中拉出」（在建構函數中），我本質上是否已經創建了一個主體？

我不確定你那句話的意思，但不是。主體（subject）同時是觀察者（observer）和可被觀察者（observable），並且具有特定的語義。僅僅像你說的那樣「把觀察者從可被觀察者中抽出」是不夠的。關於主體的語義，可以參考這裡：不同 RxJS 主體的語義是什麼？