按时间戳排序RXJS流及动态支持多流的实现方案

一、单个流的排序与逐个发射 #

你的需求是让每个流无需等待整个数据集加载完成，就能按时间戳顺序逐个发射元素。由于数据源本身已按时间戳排序，我们只需将一次性发射的数组流拆分为逐个元素的流即可；若需兼容可能乱序的数据源，可额外添加排序步骤。

修改后的dataA$和dataB$代码如下：

import { delayWhen, of, timer, from, map, concatMap } from "rxjs";

// （原dataA、dataB定义不变）

const dataA$ = of(dataA).pipe(
  delayWhen(() => timer(Math.random() * 5000)),
  // 可选：确保数组按时间戳排序（若数据源保证有序可省略）
  map(arr => [...arr].sort((a, b) => a.timestamp - b.timestamp)),
  // 将数组拆分为逐个发射的流，无需等待整个数组处理完成
  concatMap(arr => from(arr))
);

const dataB$ = of(dataB).pipe(
  delayWhen(() => timer(Math.random() * 5000)),
  map(arr => [...arr].sort((a, b) => a.timestamp - b.timestamp)),
  concatMap(arr => from(arr))
);

// 修正：每个订阅维护独立的lastTimestamp，避免跨流干扰
let lastTimestampA = -Infinity;
dataA$.subscribe(({ timestamp }) => {
  expect(timestamp > lastTimestampA).toBe(true);
  lastTimestampA = timestamp;
});

let lastTimestampB = -Infinity;
dataB$.subscribe(({ timestamp }) => {
  expect(timestamp > lastTimestampB).toBe(true);
  lastTimestampB = timestamp;
});

关键逻辑说明： #

• map(arr => [...arr].sort(...))：创建数组副本并按时间戳排序，避免修改原数据（可选步骤）。
• concatMap(arr => from(arr))：将数组转换为逐个发射元素的Observable，concatMap保证元素按原数组顺序发射，且无需等待整个数组处理完成。

二、动态支持任意数量数据流 #

分两种场景实现：

场景1：多个流独立保持内部顺序，统一管理 #

若只需每个流自身按顺序发射，可通过主题（Subject）动态维护流集合，再合并所有流：

import { delayWhen, of, timer, from, map, concatMap, BehaviorSubject, mergeAll } from "rxjs";

// 通用流处理函数，复用单个流的逻辑
const createSortedStream = (dataArray) => 
  of(dataArray).pipe(
    delayWhen(() => timer(Math.random() * 5000)),
    map(arr => [...arr].sort((a, b) => a.timestamp - b.timestamp)),
    concatMap(arr => from(arr))
  );

// 用BehaviorSubject动态维护流集合
const streamCollection$ = new BehaviorSubject([]);
const mergedStreams$ = streamCollection$.pipe(mergeAll());

// 初始添加流
streamCollection$.next([
  createSortedStream(dataA),
  createSortedStream(dataB)
]);

// 后续可动态添加新流（比如新增dataC）
// streamCollection$.next([...streamCollection$.getValue(), createSortedStream(dataC)]);

// 订阅所有流的发射（每个流内部保持顺序）
mergedStreams$.subscribe(item => {
  console.log(`发射元素：${item.data}，时间戳：${item.timestamp}`);
});

场景2：合并所有流为全局按时间戳排序的流 #

若需将多个有序流合并为一个全局有序的流，可基于归并排序思想实现：

import { delayWhen, of, timer, from, map, concatMap, merge, scan, filter } from "rxjs";

// 通用流处理函数
const createSortedStream = (dataArray) => 
  of(dataArray).pipe(
    delayWhen(() => timer(Math.random() * 5000)),
    map(arr => [...arr].sort((a, b) => a.timestamp - b.timestamp)),
    concatMap(arr => from(arr))
  );

// 合并多个有序流为全局有序流的函数
const mergeToGlobalSortedStream = (...streams) => {
  return merge(...streams.map(stream => 
    stream.pipe(map(item => ({ source: stream, item })))
  )).pipe(
    // 用scan维护每个流的待发射元素队列
    scan((acc, { source, item }) => {
      // 将当前元素加入对应流的队列
      acc.set(source, [...(acc.get(source) || []), item]);
      
      // 找到所有队列中时间戳最小的元素
      let minTimestamp = Infinity;
      let targetSource = null;
      acc.forEach((queue, src) => {
        if (queue.length && queue[0].timestamp < minTimestamp) {
          minTimestamp = queue[0].timestamp;
          targetSource = src;
        }
      });

      // 取出最小元素准备发射
      if (targetSource) {
        const emitItem = acc.get(targetSource).shift();
        // 队列空则移除该流的记录
        if (!acc.get(targetSource).length) acc.delete(targetSource);
        return { acc, emit: emitItem };
      }
      return { acc, emit: null };
    }, new Map()),
    // 过滤掉无需发射的空值
    filter(res => res.emit !== null),
    // 提取最终要发射的元素
    map(res => res.emit)
  );
};

// 创建多个流并合并
const streamA$ = createSortedStream(dataA);
const streamB$ = createSortedStream(dataB);
const globalSorted$ = mergeToGlobalSortedStream(streamA$, streamB$);

// 验证全局排序
let globalLastTimestamp = -Infinity;
globalSorted$.subscribe(({ timestamp }) => {
  expect(timestamp > globalLastTimestamp).toBe(true);
  globalLastTimestamp = timestamp;
  console.log(`全局有序发射：${timestamp}`);
});

内容的提问来源于stack exchange，提问作者Josh C.