如何从热Observable创建急切Observable？及改写操作符使序列以1开头

嘿，针对你给出的RxJS代码场景，我来帮你解决这两个问题：

问题1：从热Observable创建急切Observable #

热Observable（比如你的Subject）默认是只有当有订阅者时，才会把值推送给订阅者；而“急切Observable”指的是不需要等待订阅，就开始执行源Observable的逻辑，甚至能缓存值给后续的订阅者。

在RxJS里，我们可以用ConnectableObservable来实现这个需求：通过publish()或publishReplay()把热Observable转为可连接的Observable，然后主动调用connect()方法，让它立即订阅源Observable，不管有没有下游订阅。

代码示例 #

import * as rxjs from 'rxjs'
const { take, publishReplay } = rxjs.operators

function awesomeOpators() { return take(1); }
const numbers$ = new rxjs.Subject<number>();

// 转为可连接Observable，同时缓存最近1个值
const eagerStart$ = numbers$.pipe(
  awesomeOpators(),
  publishReplay(1)
) as rxjs.ConnectableObservable<number>;

// 主动启动，无需等待订阅
eagerStart$.connect();

numbers$.next(1);
// 此时订阅依然能拿到之前发射的1
eagerStart$.subscribe(val => console.log(val)); // 输出1
numbers$.next(2);

解释下：publishReplay(1)会帮我们缓存最近1个值，确保后续订阅的观察者也能拿到之前发射的内容；connect()调用后，Observable就开始工作了，完全不依赖下游的订阅时机。

问题2：改写awesomeOpators让start$序列以1开头 #

你原代码里的问题是：numbers$.next(1)在start$.subscribe()之前执行，但Subject不会缓存历史值，所以订阅后只能拿到之后的2。要让start$序列以1开头，分两种常见场景：

场景1：固定在序列开头插入1 #

不管源Observable有没有发射过值，都先给订阅者推一个1，再推后续的数值。用startWith操作符就能轻松实现，记得调整take的数量（原take(1)只会取第一个值，也就是1，如果想同时拿到后续的值，需要修改take参数）：

import * as rxjs from 'rxjs'
const { take, startWith } = rxjs.operators

// 改写后的函数
function awesomeOpators() { 
  return rxjs.pipe(
    startWith(1), // 在序列最前面插入1
    take(2) // 取前2个值：1和后续的2
  ); 
}

// 测试代码
const numbers$ = new rxjs.Subject<number>();
const start$ = numbers$.pipe(awesomeOpators());
numbers$.next(1);
start$.subscribe((val) => { 
  // 依次输出1、2
  console.log(val) 
});
numbers$.next(2);

场景2：回溯之前发射的1 #

如果是希望即使在numbers$.next(1)之后订阅，也能拿到这个历史值作为序列开头，就用publishReplay配合refCount来缓存历史值：

import * as rxjs from 'rxjs'
const { take, publishReplay, refCount } = rxjs.operators

// 改写后的函数
function awesomeOpators() { 
  return rxjs.pipe(
    take(1),
    publishReplay(1), // 缓存最近1个值
    refCount() // 自动管理订阅的连接与断开
  ); 
}

// 测试代码
const numbers$ = new rxjs.Subject<number>();
const start$ = numbers$.pipe(awesomeOpators());
numbers$.next(1);
start$.subscribe((val) => { 
  // 输出1
  console.log(val) 
});
numbers$.next(2);

这样不管你什么时候订阅start$，都会先拿到1作为序列的第一个值。

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