关于RxJS实现带最小与最大容量的Buffer操作符的技术问询

嘿，作为RxJS新手，你想要的这个自定义buffer操作符确实挺实用的——它结合了逐步扩容的缓冲和滑动窗口的特性，完全匹配你描述的需求。我来一步步帮你实现它～

先明确你的核心需求 #

你要的buffer(minSize, maxSize, startEvery)需要做到：

• 当缓冲的数据量达到minSize时，第一次输出当前缓冲内容
• 之后每新增一条数据，只要还没到maxSize，就输出当前全部缓冲内容
• 当缓冲达到maxSize后，每次按startEvery的步长滑动窗口，保持窗口大小为maxSize并输出

自定义操作符实现（RxJS 6语法） #

我们可以用RxJS内置的scan、filter、map等操作符组合出这个逻辑，避免从头写Observable创建逻辑，更符合RxJS的风格：

import { Observable, OperatorFunction } from 'rxjs';
import { scan, filter, map } from 'rxjs/operators';

function buffer<T>(minSize: number, maxSize: number, startEvery: number): OperatorFunction<T, T[]> {
  // 先做参数合法性校验
  if (minSize > maxSize) {
    throw new Error('minSize 不能大于 maxSize');
  }
  if (startEvery < 1) {
    throw new Error('startEvery 必须大于等于1');
  }

  return (source) => {
    return source.pipe(
      // 用scan维护缓冲状态：当前窗口、是否输出、是否满窗口、滑动起始索引
      scan((acc, value) => {
        const newWindow = [...acc.currentWindow, value];
        const windowLength = newWindow.length;

        // 情况1：窗口还没到maxSize，逐步扩容输出
        if (windowLength <= maxSize) {
          return {
            currentWindow: newWindow,
            shouldEmit: windowLength >= minSize, // 达到minSize才允许输出
            isFull: windowLength === maxSize,
            startIndex: 0
          };
        } 
        // 情况2：窗口超过maxSize，按步长滑动
        else {
          const newStartIndex = acc.startIndex + startEvery;
          // 滑动后截取窗口，再加入新值
          const slidWindow = [...acc.currentWindow.slice(newStartIndex), value];
          return {
            currentWindow: slidWindow,
            shouldEmit: true, // 满窗口后每次滑动都输出
            isFull: true,
            startIndex: newStartIndex
          };
        }
      }, {
        currentWindow: [] as T[],
        shouldEmit: false,
        isFull: false,
        startIndex: 0
      }),
      // 只保留需要输出的状态
      filter(acc => acc.shouldEmit),
      // 提取最终的缓冲数组
      map(acc => acc.currentWindow)
    );
  };
}

测试你的示例场景 #

用你给出的源Observable来测试：

import { from } from 'rxjs';

// 模拟源发出1-8的序列
const source = from([1,2,3,4,5,6,7,8]);

source.pipe(buffer(2, 4, 1)).subscribe(res => console.log(res));

运行后会输出你预期的结果：

[1,2]
[1,2,3]
[1,2,3,4]
[2,3,4,5]
[3,4,5,6]
[4,5,6,7]
[5,6,7,8]

代码关键点解释 #

• scan操作符：用来维护当前的缓冲状态，包括当前窗口数组、是否应该输出、是否达到最大容量、滑动的起始索引，每次新值进来都会更新这个状态。
• 逐步扩容阶段：当窗口长度在minSize到maxSize之间时，只要达到minSize，每新增一个值就输出当前完整窗口。
• 滑动窗口阶段：当窗口超过maxSize后，按照startEvery的步长截取旧窗口，再加入新值形成新窗口，每次滑动后都输出。
• 参数校验：提前拦截不合法的参数，避免运行时出现奇怪的行为。

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