如何用RxJS风格从单个Observable拆出两个Observable（无判断）

RxJS 原生并没有提供你伪代码中描述的那种 split 操作符，但你可以通过自定义操作符轻松实现符合你需求的“流内拆分”逻辑，完全不用显式条件判断，也能在 pipe 链内完成分支处理。

实现方式：自定义 split 操作符 #

这个操作符会根据你传入的判断函数，将原流拆分为两个子流，然后把这两个子流作为数组传递给后续操作符（比如 tap），用法和你给出的伪代码几乎一致：

import { Observable, OperatorFunction, partition, of } from 'rxjs';

// 自定义split操作符
function split<T>(predicate: (value: T) => boolean): OperatorFunction<T, [Observable<T>, Observable<T>]> {
  return (source) => {
    // 用partition拆分原流
    const [passStream, failStream] = partition(source, predicate);
    // 将两个子流包装成Observable数组输出
    return of([passStream, failStream]);
  };
}

使用示例 #

现在你可以像伪代码那样调用这个操作符：

$someObserver.pipe(
  split(foo => foo === 'Bar'),
  tap([streamA, streamB] => {
    streamA.subscribe(x => { /* Do A */ });
    streamB.subscribe(x => { /* Do B */ });
  })
).subscribe();

注意事项 #

• 如果你需要管理内部订阅的生命周期（比如防止内存泄漏），可以结合 takeUntil 等操作符，将内部订阅和主流的生命周期绑定。
• 如果你不想自定义操作符，也可以用 groupBy 实现类似逻辑，但 groupBy 本质是按分组键拆分，若仅需二分分支，自定义 split 会更简洁。

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