RxJS实现基于元素时长的数组无限循环流（解决栈溢出）

问题核心是你原来的实现中，from(data).pipe(repeat())会同步无限发射数据，没有异步延迟的情况下，RxJS会一直同步执行，最终导致调用栈溢出。要保持基于上一个元素的延迟逻辑同时无限循环，需要让整个序列是异步且按需触发的，而非同步无限生成。

最优实现方式：基于concatMap+repeat的序列循环 #

这种方式先遍历整个数组，每个元素按指定延迟发射，完成后重复整个序列，完全匹配需求：

import { from, of } from "rxjs";
import { concatMap, delay, repeat } from "rxjs/operators";

interface Entry {
  name: string;
  duration: number;
}
const data: Array<Entry> = [
  {"name": "foo", "duration": 1000},
  {"name": "bar", "duration": 4000},
  {"name": "bla", "duration": 2000}
];

// 核心逻辑：遍历数组每个元素，延迟后发射，整个序列完成后重复
const source = from(data).pipe(
  concatMap(entry => of(entry).pipe(delay(entry.duration))),
  repeat() // 无限重复整个数组的发射序列
);

source.subscribe(x => console.log(x, new Date()));

原代码栈溢出的原因 #

原来的from(data).pipe(repeat())是同步无限流：from(data)会同步发射数组所有元素，repeat()会立刻重新订阅，导致RxJS不断同步发射元素，没有任何异步操作让出调用栈，最终触发RangeError: Maximum call stack size exceeded。加上take(10)时，流会在10次发射后终止，所以不会触发栈溢出。

备选实现：用expand操作符递归生成流 #

如果需要更灵活的控制（比如动态修改数组），可以用expand操作符递归发射下一个元素：

import { of } from "rxjs";
import { delay, expand } from "rxjs/operators";

interface Entry {
  name: string;
  duration: number;
}
const data: Array<Entry> = [
  {"name": "foo", "duration": 1000},
  {"name": "bar", "duration": 4000},
  {"name": "bla", "duration": 2000}
];

// 从第一个元素开始，递归发射下一个（循环数组）
const source = of(data[0]).pipe(
  expand((_, index) => {
    const nextIndex = (index + 1) % data.length;
    return of(data[nextIndex]).pipe(delay(data[nextIndex].duration));
  }),
  delay(data[0].duration) // 第一个元素的延迟
);

source.subscribe(x => console.log(x, new Date()));

效果验证 #

两种实现都会输出类似如下的结果，并且无限循环直到取消订阅：

{ name: 'foo', duration: 1000 } 2023-10-03T20:54:37.284Z
{ name: 'bar', duration: 4000 } 2023-10-03T20:54:38.289Z
{ name: 'bla', duration: 2000 } 2023-10-03T20:54:42.294Z
{ name: 'foo', duration: 1000 } 2023-10-03T20:54:44.296Z
...

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