RxJS中mergeAll延迟订阅Observable无发射值问题求助

原代码 #

var notifySubj = new Subject<Observable<number>>();
var notify$ = notifySubj.pipe(mergeAll());
const saveSubj = new Subject<number>();
const saveAction$ = saveSubj.pipe(
  mergeMap((n) => fromFetch('https://jsonplaceholder.typicode.com/todos/' + n)),
  shareReplay({ bufferSize: 1, refCount: true })
);
const temp$: Observable<number> = saveAction$.pipe(map((r) => 1));
// notify$.subscribe((e) => console.log(e));  when placed here it emits
notifySubj.next(temp$);

saveAction$.subscribe((x) => console.log(x));
notify$.subscribe((e) => console.log(e)); // no notifications
// even when emission after subscribe
saveSubj.next(3);
saveSubj.next(4);
saveSubj.next(5);

问题描述 #

订阅notify$后调用saveSubj.next()无法收到任何值，需满足：

• 支持延迟订阅notify$
• 不接收订阅前发射的任何值

问题原因 #

temp$在saveAction$有订阅者之前就已创建并传入notifySubj，后续notify$订阅时，temp$无法关联到saveAction$的后续发射逻辑。同时shareReplay({ bufferSize:1, refCount:true })的缓存特性不符合“不接收订阅前值”的要求，提前绑定的temp$也无法感知saveAction$的订阅状态变化。

解决方案 #

方案1：用函数包装Observable延迟创建

将传入notifySubj的内容改为返回Observable的函数，确保notify$订阅时才创建temp$，绑定最新的saveAction$发射：

var notifySubj = new Subject<() => Observable<number>>();
var notify$ = notifySubj.pipe(mergeMap(fn => fn()));
const saveSubj = new Subject<number>();
const saveAction$ = saveSubj.pipe(
  mergeMap((n) => fromFetch('https://jsonplaceholder.typicode.com/todos/' + n)),
  share({ refCount: true }) // 无需缓存，用share替代shareReplay
);

// 传入函数而非直接传入Observable
notifySubj.next(() => saveAction$.pipe(map((r) => 1)));

saveAction$.subscribe((x) => console.log(x));
notify$.subscribe((e) => console.log(e));

saveSubj.next(3);
saveSubj.next(4);
saveSubj.next(5);

方案2：用defer延迟Observable实例化

通过defer延迟temp$的创建，直到notify$订阅时才绑定saveAction$：

var notifySubj = new Subject<Observable<number>>();
var notify$ = notifySubj.pipe(mergeAll());
const saveSubj = new Subject<number>();
const saveAction$ = saveSubj.pipe(
  mergeMap((n) => fromFetch('https://jsonplaceholder.typicode.com/todos/' + n)),
  share({ refCount: true })
);

// 用defer延迟创建temp$，避免提前绑定
const temp$ = defer(() => saveAction$.pipe(map((r) => 1)));
notifySubj.next(temp$);

saveAction$.subscribe((x) => console.log(x));
notify$.subscribe((e) => console.log(e));

saveSubj.next(3);
saveSubj.next(4);
saveSubj.next(5);

关键说明 #

• 替换shareReplay为share({ refCount: true })，避免缓存订阅前的值，符合需求。
• 两种方案核心都是延迟temp$与saveAction$的绑定时机，确保notify$订阅时，temp$能捕获后续所有saveSubj的发射值。

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