如何解决RxJS combineLatest与ReplaySubject的不一致行为？

问题分析 #

你遇到的combineLatest行为差异源于ReplaySubject的重放顺序与订阅时的源顺序绑定：

• 订阅[r1, r2]时，r1先重放所有历史值，r2再重放，此时r1的最新值已是r1.2，因此仅输出一次组合结果。
• 订阅[r2, r1]时，r2先重放r2.1，随后r1依次重放r1.1和r1.2，每一次重放都会触发与r2最新值的组合，因此输出两次结果。

你需要的combineEvery核心是：所有源就绪后，每个源的每一个历史值（包括后续新值）都要与其他源的最新值组合输出，且结果不受源顺序影响。

实现方案 #

以下是基于RxJS的combineEvery实现：

import { Observable, ReplaySubject, combineLatest, merge, map, switchMap, take, distinctUntilChanged } from 'rxjs';

function combineEvery<T extends readonly Observable<any>[]>(sources: T): Observable<{ [K in keyof T]: Awaited<T[K]> }> {
  // 为每个源创建ReplaySubject，缓存所有历史值
  const replaySubjects = sources.map(source => {
    const subject = new ReplaySubject<any>();
    source.subscribe(subject);
    return subject;
  });

  // 等待所有源都至少发出一个值，确保组合的合法性
  const allReady$ = combineLatest(replaySubjects).pipe(take(1));

  return allReady$.pipe(
    switchMap(() => {
      // 将每个源的流包装为带索引的对象，标记值来自哪个源
      const indexedStreams = replaySubjects.map((subject, index) => 
        subject.pipe(map(value => ({ index, value })))
      );

      // 合并所有源的流，处理每个值的组合逻辑
      return merge(...indexedStreams).pipe(
        map(({ index, value }) => {
          // 构建组合数组：当前源用当前值，其他源取最新缓存值
          return replaySubjects.map((subject, i) => 
            i === index ? value : subject.getValue()
          ) as { [K in keyof T]: Awaited<T[K]> };
        }),
        // 去重，避免同一组合重复输出
        distinctUntilChanged((prev, curr) => JSON.stringify(prev) === JSON.stringify(curr))
      );
    })
  );
}

测试验证 #

双源场景测试 #

const r1 = new ReplaySubject(2);
const r2 = new ReplaySubject(2);
r1.next('r1.1');
r1.next('r1.2');
r2.next('r2.1');

// 无论源顺序如何，输出结果一致
combineEvery([r1, r2]).subscribe(console.log);
// 输出：
// ["r1.1", "r2.1"]
// ["r1.2", "r2.1"]

combineEvery([r2, r1]).subscribe(console.log);
// 输出：
// ["r2.1", "r1.1"]
// ["r2.1", "r1.2"]

三源场景测试 #

const r1 = new ReplaySubject();
const r2 = new ReplaySubject();
const r3 = new ReplaySubject();

r1.next('r1.1');
r1.next('r1.2');
r2.next('r2.1');
r3.next('r3.1');

combineEvery([r1, r2, r3]).subscribe(console.log);
// 输出：
// ["r1.1", "r2.1", "r3.1"]
// ["r1.2", "r2.1", "r3.1"]

实现逻辑说明 #

1. 缓存历史值：用ReplaySubject包裹每个输入源，确保能获取所有已发出的历史值。
2. 等待就绪：通过combineLatest取所有源的第一个值，确保所有源都有可组合的数据。
3. 合并处理：将所有源的流合并，每个值携带源索引，构建组合时当前源用自身值，其他源用最新缓存值。
4. 去重优化：用distinctUntilChanged避免因重复触发导致的相同组合输出。

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