RxJS 动态变化数组的流发射与计算切换实现咨询

• 数字型源数组随机触发push新增、pop删除操作
• 计算任务执行约束：上一轮计算完全执行结束后，才会启动下一轮计算流程
• 数组变更触发的特殊规则：只要数组发生变化，立刻丢弃所有未完成的旧计算任务，基于最新的数组值重新启动全量计算
• 现有实现的待解决问题：switchMap内部传递给of的数组存在类型识别异常，不清楚如何适配from操作符优化逻辑

let nums: number[] = []

const nums$ = new Subject()

timer(2000, 5000).pipe(
    concatMap(n => of(n).pipe(delay(Math.random() * 1000)))
).subscribe(val => {
    nums.push(val)
    nums$.next(nums)
})

nums$.pipe(
    switchMap(n => of(n).pipe(delay(1000)))
).subscribe(val => {
    console.log(val)
    nums$.next(nums)
})

1. 初始化nums$时未指定泛型，默认推导出Subject<unknown>类型，导致下游管道拿到的参数没有类型信息，传给of、from这类操作符时自然无法识别数组结构
2. 逻辑冗余：计算完成后主动触发nums$.next(nums)的代码会无意义重复推送同一份数组数据，既没有真实变更事件作为触发源，还容易造成循环调用的隐患
3. 行为差异适配：of(nums)是直接把整个数组作为单个值发射，from(nums)是把数组拆解为单个元素依次发射，直接替换会改变计算的输出节奏，需要搭配延时逻辑适配原来模拟计算耗时的需求

核心调整方向：显式补全泛型解决类型问题，去掉冗余触发逻辑，适配from的行为模拟计算流程

import { Subject, timer, of, from, concatMap, switchMap, delay } from 'rxjs'

let nums: number[] = []
// 显式指定泛型，彻底解决类型识别问题
const nums$ = new Subject<number[]>()

// 模拟数组随机变更逻辑：随机触发新增/删除
timer(2000, 5000).pipe(
    concatMap(n => of(n).pipe(delay(Math.random() * 1000)))
).subscribe(() => {
    // 非空数组时50%概率执行pop删除
    if (Math.random() > 0.5 && nums.length > 0) {
        nums.pop()
    } else {
        nums.push(Math.floor(Math.random() * 100))
    }
    // 数组变更后推送最新值
    nums$.next(nums)
})

// 计算管道
nums$.pipe(
    switchMap(currentNums => 
        // 用from拆解数组元素，配合delay模拟计算耗时
        from(currentNums).pipe(
            delay(1000)
        )
    )
).subscribe(val => {
    console.log('计算输出:', val)
})

• switchMap 天然满足「数组变更即丢弃未完成旧计算」的要求：每次源流推送新数组时，会立刻退订上一个未跑完的内部计算流
• 内部计算通过from按顺序遍历数组元素，每个元素带1s延时模拟计算耗时，所有元素处理完才算一轮计算完成，刚好满足「上一次计算结束才进入下一轮处理」的约束
• 显式指定的泛型会被TS链路正确识别，不存在数组类型匹配失败的问题

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