如何在不重复订阅的情况下将Observable与Promise共享？

在Angular环境中使用Observable时，希望让Observable和Promise共享同一个值。由于服务在SSR（服务器端渲染）场景下不会等待第一个值发出后再解析，因此需要Promise版本。当前代码存在重复订阅Observable的问题，想知道如何在不重复订阅的情况下实现值共享，是否可以使用share()或shareReplay()操作符，或者有没有办法不用创建两个订阅就能等待第一个值发出。

当前代码：

export class Numbers implements OnDestroy {

    private _num: Numbers | null;
    private _numObs: Observable<Numbers | null>;
    private _numSub!: Subscription;

    constructor(
        private ns: NumberService
    ) {
        // initialize as null
        this._num = null;

        // the shared observable
        this._numObs = this.ns.numbers$;

        // subscribe to the observable, save state
        this._numSub = this._numObs.subscribe(_num => {
            this._num = _num;
        });
    }

    async toPromise(): Promise<Numbers | null> {
        return await firstValueFrom(this._numObs);
    }

    get numbers(): Numbers | null {
        // get state
        return this._num;
    }

    ngOnDestroy(): void {
        this._numSub.unsubscribe();
    }
}

核心是把Observable改成多播流，让所有订阅者共用同一个源订阅，避免重复触发上游逻辑。结合你的SSR场景，用shareReplay(1)是最优选择——它既能实现订阅共享，还能缓存最新的1个值，刚好适配SSR需要获取已有值的需求。

具体改造步骤 #

1. 包装原Observable：初始化共享Observable时，给原流添加shareReplay(1)操作符，这样后续所有订阅都会复用同一个源，且能拿到最新的已发出值。
2. 保留单次订阅维护状态：构造函数里的订阅用来同步更新本地_num状态，这个订阅和toPromise里的调用会共享同一个源，不会重复触发上游逻辑。

优化后的代码：

import { firstValueFrom, Observable, shareReplay, Subscription } from 'rxjs';
import { OnDestroy } from '@angular/core';
import { NumberService } from './number.service';

export class Numbers implements OnDestroy {
    private _num: Numbers | null = null;
    private _sharedNumObs: Observable<Numbers | null>;
    private _subscription!: Subscription;

    constructor(private ns: NumberService) {
        // 将原Observable转为多播且缓存最新值的流
        this._sharedNumObs = this.ns.numbers$.pipe(
            shareReplay(1) // 缓存1个最新值，所有订阅共享同一个源订阅
        );

        // 仅订阅一次，同步维护本地状态
        this._subscription = this._sharedNumObs.subscribe(num => {
            this._num = num;
        });
    }

    async toPromise(): Promise<Numbers | null> {
        // 此处订阅会复用上面的源，不会重复触发上游逻辑
        return firstValueFrom(this._sharedNumObs);
    }

    get numbers(): Numbers | null {
        return this._num;
    }

    ngOnDestroy(): void {
        this._subscription.unsubscribe();
    }
}

为什么选shareReplay(1)而非share()？ #

• share()仅能将冷Observable转为热Observable，但如果源已经发出过值，新订阅无法获取历史值——这在SSR场景下可能导致Promise拿不到已存在的值，直接阻塞。
• shareReplay(1)会缓存最新的1个值，无论何时订阅（包括SSR时的Promise调用），都能立即拿到最新值，完美匹配你的需求。

更简洁的可选写法 #

如果不需要同步的numbers getter，或者可以接受getter返回Promise，还能进一步简化：

export class Numbers implements OnDestroy {
    private _sharedNumObs: Observable<Numbers | null>;
    private _subscription!: Subscription;

    constructor(private ns: NumberService) {
        this._sharedNumObs = this.ns.numbers$.pipe(shareReplay(1));
        // 若不需要同步维护本地状态，可去掉该订阅，让firstValueFrom自动触发源
        this._subscription = this._sharedNumObs.subscribe();
    }

    async toPromise(): Promise<Numbers | null> {
        return firstValueFrom(this._sharedNumObs);
    }

    // 如需同步获取值，可使用getValue，但要确保Observable已发出过值
    get numbers(): Numbers | null {
        try {
            return this._sharedNumObs.pipe(take(1)).getValue();
        } catch {
            return null;
        }
    }

    ngOnDestroy(): void {
        this._subscription.unsubscribe();
    }
}

关键注意事项 #

• SSR环境下，要确保NumberService的numbers$在服务器端能正常发出值，否则shareReplay(1)没有缓存内容，Promise仍无法获取值。
• 务必在ngOnDestroy中取消订阅，避免内存泄漏。

内容的提问来源于stack exchange，提问作者Jonathan