如何将冷Observable转为热Observable？RxJS代码优化求助

问题分析 #

你遇到的核心问题是take(1)和share()的默认行为冲突了：

• take(1)会在流发出第一个值后立即完成Observable，同时自动取消当前订阅。
• share()默认使用refCount: true，当所有订阅者都取消订阅（这里因为take(1)完成导致），refCount降到0时，它会断开对源Observable的订阅。当新的订阅者进来时，share()会重新订阅源，这就导致你的tap日志被执行了5次（对应5个订阅）。

解决方案 #

这里有两种常用的修复方式，根据你的需求选择：

方案1：将take(1)移到订阅环节（推荐）

把take(1)从共享流中移除，让每个订阅者自己处理“只取第一个值”的逻辑，这样共享流不会被提前完成，share()能正常保持源的订阅状态：

const test = new BehaviorSubject('test').pipe(tap(() => console.log('I want this to be logged only once to the console!')))
const grr = test.pipe(share())

// 每个订阅者单独添加take(1)
grr.pipe(take(1)).subscribe(() => console.log(1))
grr.pipe(take(1)).subscribe(() => console.log(2))
grr.pipe(take(1)).subscribe(() => console.log(3))
grr.pipe(take(1)).subscribe(() => console.log(4))
grr.pipe(take(1)).subscribe(() => console.log(5))

方案2：使用shareReplay替代share()

如果你一定要把take(1)放在共享流里，可以用shareReplay(1)。它会缓存最近的1个值，并且即使源Observable完成，新订阅也会直接拿到缓存的值，不会重新订阅源：

const test = new BehaviorSubject('test').pipe(tap(() => console.log('I want this to be logged only once to the console!')))
const grr = test.pipe(
  shareReplay(1), // 替换share()为shareReplay
  take(1)
)

grr.subscribe(() => console.log(1))
grr.subscribe(() => console.log(2))
grr.subscribe(() => console.log(3))
grr.subscribe(() => console.log(4))
grr.subscribe(() => console.log(5))

或者你也可以调整share()的配置，禁用完成后的重置：

const grr = test.pipe(
  share({ resetOnComplete: false }),
  take(1)
)

验证结果 #

两种方案都会输出你期望的结果：

'I want this to be logged only once to the console!'
1
2
3
4
5

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