如何按除以2余数规则对数字Observable执行window条件分组

你要的是连续同规则分组，不是全局按奇偶归类：只要当前元素和前一个元素除以2的余数不同，就立刻切新窗口。核心是要维护相邻元素的余数状态作为窗口切换的判断依据，注意不要用groupBy——那个操作符会把序列里所有同余数的元素不管位置都归到同一个组，输出会变成[1,3,5,1,3]、[2,4]，不符合连续分组的要求。

可直接运行的实现代码 #

const { from, Subject, merge } = require('rxjs');
const { window, scan, lastValueFrom } = require('rxjs/operators');

// 源数字序列
const source$ = from([1, 3, 5, 2, 4, 1, 3]);
// 用来发送窗口关闭/新窗口开启信号的通知流
const windowBoundary$ = new Subject();

// 扫描跟踪相邻元素的余数变化，余数切换时发边界信号
source$.pipe(
  scan((lastMod, currentVal) => {
    const currentMod = currentVal % 2;
    if (lastMod !== null && lastMod !== currentMod) {
      windowBoundary$.next();
    }
    return currentMod;
  }, null)
).subscribe(); // 订阅触发扫描逻辑，否则状态跟踪不会执行

// 组装最终的高阶Observable
const groupedWindow$ = source$.pipe(
  window(merge(windowBoundary$))
);

// 测试：订阅输出每个窗口的内容
groupedWindow$.subscribe(async (singleWindow$) => {
  const windowContent = await lastValueFrom(singleWindow$.pipe(
    scan((arr, val) => [...arr, val], [])
  ));
  console.log(windowContent);
});

运行后输出顺序和预期完全一致：

[1, 3, 5]
[2, 4]
[1, 3]

实现逻辑说明 #

• 用scan维护上一个元素除以2的余数值，遍历每个元素时和当前余数对比，不一致就发送窗口切换信号
• window操作符收到切换信号就会关闭当前正在推送数据的窗口Observable，同时开启新的窗口Observable承接后续元素
• 第一个窗口不需要额外触发开启：源序列发射第一个元素时window会自动创建初始窗口，符合连续分组的起始逻辑
• 最终输出就是要求的高阶Observable，每个内层Observable对应一组连续同余的数字序列，可按需对每个内层流做后续运算。

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