如何让预订阅共享的RxJS Observable具备Promise特性，无需额外subscribe()？

要让RxJS Observable具备Promise的「创建即执行、缓存结果、后续订阅直接获取结果」特性，我们可以通过封装一个工具函数，结合publishReplay和connect操作符来实现，完全不需要手动调用subscribe()。

核心解决方案 #

我们可以写一个工具函数，把任意冷Observable转换成预订阅的热Observable，同时缓存它的所有输出（包括值、错误和完成信号）：

import { Observable, ConnectableObservable } from 'rxjs';
import { publishReplay } from 'rxjs/operators';

function preSubscribed<T>(source: Observable<T>): Observable<T> {
  // 将源Observable转为可连接的Observable，并缓存所有发出的值
  const connectable = source.pipe(publishReplay()) as ConnectableObservable<T>;
  // 立即触发订阅，让源Observable在创建时就执行
  connectable.connect();
  return connectable;
}

为什么这个方案有效？ #

• publishReplay()：把冷Observable包装成ConnectableObservable，它会缓存源Observable发出的所有值（你也可以通过参数指定缓存数量，比如publishReplay(1)只缓存最新值），后续订阅者会直接拿到缓存的结果，不会重新触发源Observable的执行。
• connect()：手动触发ConnectableObservable对源Observable的订阅，这一步让Observable在创建时就立即执行，完全不需要外部调用subscribe()。
• 缓存特性：不管源Observable是正在执行还是已经完成，后续的订阅都能获取到之前缓存的结果，完美复现Promise的行为。

示例验证 #

示例1：基础场景 #

import { of } from 'rxjs';
import { tap, map } from 'rxjs/operators';

const obs = preSubscribed(
  of(1).pipe(
    tap(() => console.log(`${Date.now()} ms map 1`)),
    map(x => x + 1)
  )
);

// 100ms后再订阅，依然能拿到缓存的结果
setTimeout(() => {
  obs.subscribe(val => console.log(`${Date.now()} ms subscribe ${val}`));
}, 100);

输出（符合理想结果）：

0 ms map 1
100 ms subscribe 2

示例2：已完成的Observable场景 #

const obs = preSubscribed(
  of(1, 2).pipe(
    tap(x => console.log(`${Date.now()} ms map, ${x}`)),
    map(x => x + 1)
  )
);

setTimeout(() => {
  obs.subscribe(val => console.log(`${Date.now()} ms subscribe ${val}`));
  obs.subscribe(val => console.log(`${Date.now()} ms subscribe ${val}`));
}, 100);

输出（符合理想结果）：

0 ms map, 1
0 ms map, 2
100 ms subscribe 2
100 ms subscribe 3

注意事项 #

• 如果需要限制缓存的数量（比如只保留最新的N个值），可以给publishReplay传参数，比如publishReplay(1)只缓存最后一个值。
• 错误处理：如果源Observable抛出错误，后续订阅也会收到这个错误，和Promise的catch行为一致。
• 资源清理：如果你需要手动终止源Observable的执行，可以保存connect()返回的订阅对象，在合适的时机调用unsubscribe()：

  const subscription = connectable.connect();
  // 后续需要清理时
  subscription.unsubscribe();
  

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