RxJS问题：如何让Subject在所有源Observable完成后再完成？

嘿，这个场景我太熟悉了！你现在的问题根源在于：当你直接把多个Observable订阅到同一个Subject上时，只要其中任意一个源Observable调用了complete()，这个Subject就会立刻触发自己的complete，导致后续其他源的事件再也发不出来。所以你才只拿到了前几个输出，没等到b的最后两次发射。

那怎么让合并后的流等所有源都完成再结束呢？其实不用手动管理Subject的complete时机，有两种简单的实现方式，看你需求选就行：

方式一：用merge操作符（推荐） #

RxJS的merge操作符天生就支持“合并多个流，且所有流都完成后才会完成”的特性，完全符合你的需求，代码也最简洁：

import { interval, merge } from 'rxjs';
import { mapTo, take } from 'rxjs/operators';

const a = interval(100).pipe(take(3), mapTo('a'));
const b = interval(150).pipe(take(3), mapTo('b'));

// merge会自动合并两个流，等所有源都完成后才触发complete
const c = merge(a, b);
c.subscribe(console.log);

方式二：手动控制Subject的complete时机（适合有自定义逻辑的场景） #

如果你一定要保留手动创建Subject的方式（比如需要在Subject里做额外的事件处理），可以通过forkJoin监听所有源的完成状态，只有当所有源都完成时，再调用Subject的complete()：

import { interval, Subject, forkJoin } from 'rxjs';
import { mapTo, take } from 'rxjs/operators';

const c = new Subject<string>();
const a = interval(100).pipe(take(3), mapTo('a'));
const b = interval(150).pipe(take(3), mapTo('b'));

// 订阅所有源的next和error事件到Subject，但不转发complete
a.subscribe({
  next: val => c.next(val),
  error: err => c.error(err)
});
b.subscribe({
  next: val => c.next(val),
  error: err => c.error(err)
});

// 用forkJoin等待所有源都完成后，再手动结束Subject
forkJoin([a, b]).subscribe(() => {
  c.complete();
});

c.subscribe(console.log);

为什么这样能解决问题？ #

• merge操作符：它会同时订阅所有传入的Observable，转发每个Observable的next事件，只有当所有源Observable都完成时，merge返回的Observable才会触发complete，完美匹配你的期望。
• 手动控制方式：forkJoin会等待所有传入的Observable都完成后才会发射值，这时我们再调用c.complete()，就能保证Subject在所有源都处理完后才结束。

运行上面的代码，你就能得到期望的输出：a b a a b b啦！

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