RxJS如何实现两个Observable交替取值的需求？

核心实现逻辑 #

你需要的交替触发效果可以通过 take(1) + concat() + repeat() 组合实现，核心是控制单次流程的取值规则，再循环执行整个流程。

代码示例 #

首先导入依赖的操作符：

import { concat, take, repeat, Subject } from 'rxjs';

// 替换为你自己的两个源Observable
const obs1$ = new Subject<number>();
const obs2$ = new Subject<string | number>();

核心逻辑代码：

const alternatingStream$ = concat(
  // 第一阶段：仅取obs1的1个值，取完自动结束当前段，不再监听obs1后续输出
  obs1$.pipe(take(1)),
  // 第二阶段：仅取obs2的1个值，取完自动结束当前段，不再监听obs2后续输出
  obs2$.pipe(take(1))
).pipe(
  // 重复执行上面的两阶段流程，实现循环交替
  repeat()
);

原理解释 #

• take(1) 保证每个Observable只取一次值就终止当前段的订阅，所以进入obs2的等待阶段时，已经没有对obs1的活跃订阅，obs1的所有输出都会被忽略。
• concat() 严格按顺序执行订阅：前一个流未完成时，不会订阅后一个流，完全符合你要求的「先等obs1→再等obs2→再等obs1」的顺序逻辑。
• repeat() 会在整个concat流程走完（即obs1、obs2各取了一次值）之后，重新订阅整个concat流，开启下一轮交替。

测试验证 #

你可以用下面的代码验证效果：

alternatingStream$.subscribe(val => {
  console.log('输出值：', val);
});

// 模拟事件触发
obs1$.next(1); // 输出：输出值：1
obs1$.next(2); // 无输出，处于等待obs2阶段，obs1输出被忽略
obs2$.next('a'); // 输出：输出值：a
obs2$.next('b'); // 无输出，回到等待obs1阶段，obs2输出被忽略
obs1$.next(3); // 输出：输出值：3
obs2$.next('c'); // 输出：输出值：c

扩展说明 #

如果需要限制交替次数，直接给repeat()传入数字参数即可，比如repeat(5)代表交替执行5轮后自动终止流。

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