RxJS如何基于自定义比较器缓冲元素 遇不同值时触发缓冲区刷新

核心思路 #

你需要的缓冲逻辑依赖前后元素的属性对比，buffer/bufferWhen这类操作符的关闭通知无法直接关联前后元素的状态，更适合用scan操作符维护缓冲区状态，配合状态标记实现按需输出缓冲。

完整可运行代码 #

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

// 示例源数据流
const source$ = from([
  { t: 10, price:12 },
  { t: 10, price:13 },
  { t: 10, price:14 },
  { t: 11, price:12 },
  { t: 11, price:13 },
  { t: 10, price:14 },
  { t: 10, price:15 },
]);

// 自定义比较器：可根据需求替换分组规则
const isSameGroup = (prev, curr) => prev.t === curr.t;

source$.pipe(
  // 把所有事件（next/complete/error）转为通知对象，方便处理完成时的剩余缓冲
  materialize(),
  // 维护缓冲区状态，emit字段标记是否需要输出当前缓冲
  scan((state, notification) => {
    // 处理完成事件：输出最后剩下的缓冲区内容
    if (notification.kind === 'C') {
      return {
        emit: state.buffer.length ? state.buffer : null,
        buffer: []
      };
    }
    const currValue = notification.value;
    // 缓冲区为空时直接添加当前元素
    if (state.buffer.length === 0) {
      return { emit: null, buffer: [currValue] };
    }
    // 和缓冲区最后一个元素比较，同组则追加
    if (isSameGroup(state.buffer[state.buffer.length - 1], currValue)) {
      return { emit: null, buffer: [...state.buffer, currValue] };
    }
    // 不同组则输出旧缓冲，用当前元素初始化新缓冲
    return { emit: state.buffer, buffer: [currValue] };
  }, { emit: null, buffer: [] }),
  // 过滤出需要输出的缓冲
  filter(state => state.emit !== null),
  map(state => state.emit)
).subscribe(console.log);

输出结果 #

[ { t: 10, price: 12 }, { t: 10, price: 13 }, { t: 10, price: 14 } ]
[ { t: 11, price: 12 }, { t: 11, price: 13 } ]
[ { t: 10, price: 14 }, { t: 10, price: 15 } ]

方案说明 #

• 自定义的isSameGroup比较器可以任意替换，适配不同的分组判断规则
• 用materialize处理完成事件，保证数据流结束时最后一组未触发输出的缓冲也能正常返回
• 缓冲数组每次生成新实例，无副作用，不会造成状态污染

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