RxJS 如何实现每批10个值的批量处理直到上游Observable完成？

你当前实现卡点的本质是混淆了两类数组聚合操作符的行为差异：

• toArray 的逻辑是缓存上游从启动到终止的全部元素，只有等整个流彻底结束才会把所有数据打包成单个大数组输出，完全不满足“攒满10个就立刻处理当前批次”的要求
• 你缺失的关键实现逻辑，是使用专门做固定数量分块的操作符 bufferCount，这个操作符的原生能力就是：每攒够指定数量的上游元素，就立刻把缓存的元素打包成数组推给下游，清空缓存后继续攒下一批，完全匹配你的批次粒度需求。

你原代码里还有一处细节疏漏：当前写的take(10)只会取10个元素，根本达不到你说的“10批*10个元素共100个数据”的总目标，调整后的可运行代码如下：

import { interval } from 'rxjs';
import { bufferCount, take, tap } from 'rxjs/operators';

interval(100)
  .pipe(
    // 取满100个上游元素，凑齐10批的总数据量
    take(100),
    // 每攒满10个元素就打包成数组输出，形成固定粒度的批次流
    bufferCount(10),
    // 此时tap每次接收到的就是长度为10的批次数组
    tap(batch => {
      // 在这里执行批次处理逻辑即可
      console.log('处理当前批次，共', batch.length, '个元素');
    })
  )
  .subscribe(batch => {
    console.log('输出收到的批次：', batch);
  });

• bufferCount(10) 不会等待整个上游流结束才输出第一批，只要缓存够10个元素就会立刻触发批次输出，完全符合你要的“边收边处理、不需要等全量数据到齐”的逻辑
• 边界兼容特性：如果上游在没攒够10个元素时就提前终止，bufferCount 会把最后剩余的不足10个的元素也打包成数组发出，不会出现数据丢失
• 如果你需要严格只保留满额10个的批次、丢弃最后不足10个的残缺数据块，可以在bufferCount之后增加长度判断过滤逻辑。

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