半年前我在本地新建了一个空目录，名字就叫rea，没有说明文档，没有 package.json，只有三个字母。后来我给它补全了一个含义：REA = Real-time Event Aggregator，一个轻量级实时事件聚合器。这算是把项目名从“名字”变成“定义”的过程。如果你正在做监控大屏、设备遥测、实时日志这类前端页面，大概率会遇到同一个问题：后端推送的数据太密、太碎，直接渲染会卡，全量存内存会爆，而业务上你真正需要的，可能只是“每秒动态、每10秒均值、异常数”这些东西。REA 就是用来做这层加工的工具。这篇文章我按项目复盘的思路来写：为什么需要它、API 是怎么设计的、核心实现里有哪些细节和坑、压测表现怎么样。里面涉及的所有代码都是我在实际开发中写过的简化版，可以直接拿去改。1. 为什么需要 REA：一个实时数据小场景1.1 你遇到过的“数据风暴”之前我在做一个设备状态面板，后端通过 MQTT 网关推送传感器数据，频率大概每秒 2000 条。每条数据长这样：{ "deviceId": "dev-001", "timestamp": 1735123456789, "temperature": 23.4, "humidity": 43.2, "status": "online" }刚开始我图省事，前端直接在onmessage里把数据push进一个数组，然后用setInterval每秒取一次数组更新图表。结果启动不到两分钟，页面就开始掉帧，DevTools 里内存曲线一路爬升到 200MB。原因很直接：5 万条数据全部堆积在内存里，图表库每次渲染都要遍历整个数组，大量 DOM 更新被阻塞。更要命的是，这种“实时数据”里超过 80% 是冗余的——同一台设备一秒钟上报 5 次温度，屏幕上的折线图根本看不出来差异。我在那一瞬间意识到，前端需要的不只是“收到就显示”，而是“把收到的东西加工成有意义的信号”之后再显示。这个加工层，就是实时事件聚合。1.2 为什么不在后端做聚合有人会问：后端直接聚合好再推给前端不行吗？可以，但现实往往不允许。后端的推送通道通常是通用数据管道，面向多个消费方——有人要原始数据做离线分析，有人要按设备聚合，有人要全量日志，让后端为每一种前端需求单独设计聚合接口，成本和沟通周期都很高。你要的是一个“滑动窗口内求均值”的口径，改一次需求就要后端改一次推送结构，联调一遍，这在快速迭代的场景里非常拖节奏。所以我把聚合逻辑放在前端，做成一个独立库，由调用方自己决定窗口大小、分组维度、统计指标。这也是 REA 核心的设计立场：原始数据不动，分析口径由消费方定义。1.3 REA 与防抖节流的本质区别很多同事第一次看到 REA 会问“这不就是 lodash 的 throttle 吗？”还真不是。防抖和节流解决的是“调用频率”问题，它们只关心一个动作多久执行一次；而聚合解决的是“数据到信息”的转换问题，它要做时间窗口、分组、去重、统计和采样。举一个例子：一批设备状态消息，1 秒内来了 2000 条。节流能做到“这 2000 条只触发一次 UI 更新”，但 UI 拿到的还是 2000 条原始数据，该卡还是卡。REA 能把这 2000 条变成“16 台设备各自的平均温度、最高湿度、在线率”，输出一个体积很小的报告。用摄影类比：防抖节流是让你别连续按快门，聚合是先把拍到的几百张照片自动筛选拼接成一张全景图，你拿到的是结果，不是底片。2. REA 的 API 设计与核心模型2.1 三句话描述 REAREA 对外暴露一个核心方法createStream，它返回一个流对象。你通过push把原始事件投进去，通过on订阅聚合后的报告，中间用window、groupBy、metrics配置来决定“怎么聚”。整个过程是一个链式管道。第一版我把它设计成纯粹的面向对象风格，每个统计器是一个类。后来发现类层次太深，调用方自定义指标时很别扭。重构后改成函数式管道：一个流就是一组小模块的串联，最后接一个订阅回调。这样既好测试，也好扩展。import { createStream } from 'rea'; interface DeviceEvent { deviceId: string; timestamp: number; temperature: number; humidity: number; status: 'online' | 'offline'; } const stream = createStream<DeviceEvent>({ aggregate: { // 每 1000ms 形成一个窗口 windowMs: 1000, // 按设备分组 groupBy: 'deviceId', metrics: { avgTemp: ['temperature', 'avg'], maxHum: ['humidity', 'max'], offlineCount: ['status', 'countIf', v => v === 'offline'] } } }); stream.on('report', (report) => { renderDeviceCards(report.groups); }); // 模拟消息入口 stream.push({ deviceId: 'dev-001', timestamp: Date.now(), temperature: 23.5, humidity: 41, status: 'online' });2.2 为什么不用现成的流式库在 REA 之前，我考虑过直接引入响应式编程库。它们功能确实完整，冷热流、背压、调度器、操作符一应俱全。但引入到项目里之后我发现自己只需要其中三个操作：按时间切片、按字段分组、算几个指标。而完整库带来的概念负担，比如 Subject、Observable、Scheduler，对于只想要一个“数据加工器”的团队来说太重了。REA 刻意缩小了范围：没有高阶流、没有异步调度编排、没有操作符组合子。流对象本质上是“一个可推送的队列加一组归约函数”，你给它数据，它按配置吐报告。世界上的问题不是都要用图灵完备的抽象来解决，一个跑步机比一台跑车更适合放在客厅里。2.3 核心 API 与配置项用表格整理一下第一版稳定 API，方便你对照自己的场景。API / 配置类型说明createStream(options)函数创建聚合流，返回 Stream 实例stream.push(event)方法推入一条原始事件stream.on('report', cb)方法订阅聚合报告，窗口结束时触发stream.start()/stream.stop()方法手动启停内部定时器windowMsnumber统计窗口时长，单位毫秒groupBystring | (event) => string分组字段或自定义分组函数metricsRecord<string, Metric>定义输出指标，metrics 内部是归约配置emitEmptyboolean窗口内无数据时是否输出空报告，默认 falsededupeboolean | (event) => string去重开关或自定义去重键函数pollIntervalnumber检查窗口边界的时间间隔，默认 100msbufferSizenumber输入缓冲队列上限，超过后按丢弃策略处理第一版没有配置项乱飞的毛病，是在使用中慢慢加出来的：emitEmpty是监控大屏需要水位线，即使没有新数据也要让 UI 保持“活着”的视觉效果；bufferSize是压测后加的，事件爆发时宁可丢不要爆。2.4 用类型把事件结构约束住实时事件数据的结构，有时候真的只有上线那一刻才知道。所以 REA 在 TypeScript 泛型上花了一些功夫：DeviceEvent这个 interface 可以完全由使用者定义，groupBy的类型会根据传入的 interface 自动推导，metrics里指定的字段如果不存在，编译期就会报错。// 错误示例：编译不通过 const stream = createStream<DeviceEvent>({ aggregate: { metrics: { invalidMetric: ['nonExistentField', 'avg'] // 报错：DeviceEvent 没有 nonExistentField } } });这算是我自己比较满意的一个设计点：让“配置写错”从运行时的静默异常，变成编译期的红波浪线。团队新人接手配置时，不容易把字段名拼错。3. 核心实现细节与调优3.1 滑动时间窗口：不是定时清数组第一时间想到的实现方案肯定是：开一个setInterval，每 1 秒把数组里的东西全算一遍，然后清空数组。这个方案的问题在于时间边界——如果一条事件的timestamp是 00:00:00.950，另一条是 00:00:01.050，按照绝对秒来切窗口，它们会被分到两个窗口。但对于一个“最近 1 秒”的滑动窗口，它们其实应该落在同一个窗口里。要正确实现滑动窗口，内存里不应该存“一个窗口的数据”，而应该存“一个时间范围内的事件”。我用的是环形缓冲加“过期指针”的方案：固定长度数组存储事件，头部指针和尾部指针分别指向最早和最晚事件；每次push时，把事件写到尾部，尾部指针后移；每次触发聚合时，从头部开始向前遍历，把timestamp超出windowMs的事件标记为过期，头部指针后移；剩下的数据就是“活窗口”。这样做的好处是，聚合操作不需要反复filter全量数组，遍历的永远只是活数据。我用的环形缓冲简化版大概是这样的：class RingBuffer<T extends { timestamp: number }> { private buffer: (T | null)[]; private head = 0; private tail = 0; private size = 0; constructor(private capacity: number) { this.buffer = new Array(capacity).fill(null); } push(item: T): boolean { if (this.size === this.capacity) { return false; // 队列满，丢弃策略由外部决定 } this.buffer[this.tail] = item; this.tail = (this.tail + 1) % this.capacity; this.size++; return true; } expireBefore(timestamp: number): void { while (this.size > 0) { const item = this.buffer[this.head]; if (!item || item.timestamp >= timestamp) break; this.buffer[this.head] = null; this.head = (this.head + 1) % this.capacity; this.size--; } } *activeItems(): Generator<T> { for (let i = 0; i < this.size; i++) { const idx = (this.head + i) % this.capacity; const item = this.buffer[idx]; if (item) yield item; } } }这个方案有两个真实好处：一是内存占用是预分配的，不会因为数据量上涨导致频繁 GC；二是“过期清理”是 O(1) 操作，不是每次聚合并遍历全量。初始容量根据业务峰值定，比如capacity = maxEventsPerSecond * 2，不够再扩容。注意：环形缓冲里的capacity要留出余量。如果每秒 2000 条、窗口 1 秒，缓冲容量至少开 4000，给事件突发和窗口重叠留空间。3.2 聚合统计：增量计算替代重复遍历窗口内 2000 条事件，如果每次报告都把所有事件重新求和再求平均，代价并不大——2000 次加法对 CPU 来说毫秒级。但当窗口变成 5 秒、事件变成每秒 1 万条，这种朴素算法的性能就不够看了。REA 在指标计算上做了两层优化：第一层，增量归约。每次push事件时，不是等窗口结束才计算，而是在事件进来时立刻更新一个聚合状态：累计值、最大值、最小值、计数。窗口结束时只需读取这个状态，O(1) 时间就能给出报告。为了让“求平均”这类指标能被增量计算而不存储原始数据，我定义了AggregateState接口：interface AggregateState { sum: number; count: number; max: number; min: number; latest: number; }每来一条事件，sum += value; count++; if (value > max) max = value，这样窗口结束直接sum / count就是平均。内存占用从“存 N 条事件”降到“存 N 个状态”。第二层，事件类型分桶。REA 内部把事件先按groupBy分组，每个分组独立维护聚合状态。这样同一个设备的事件只会更新属于自己的那一个状态，不会出现所有设备混在一个计数器里的口径错误。3.3 分帧处理：把 2000 条推送掰成几次小任务即使聚合计算再快，如果push是在同一个事件循环 tick 里连续处理 2000 次，UI 线程仍然会被长时间占用。这里的核心问题是：数据进入的速度不应该完全决定处理的节奏。REA 采用了一种“分帧批处理”的策略：push只做入队动作，复杂度 O(1)；聚合计算由一个调度器触发，默认每pollInterval（100ms）处理一次队列里的所有事件，但如果一次要处理的事件超过阈值（比如 500 条），就把这批事件拆到多个宏任务里处理，让浏览器有时间插队渲染。// 伪代码：分帧调度 let pendingCount = 0; const MAX_BATCH = 500; function flushQueue() { const batch = queue.dequeueUpTo(MAX_BATCH); if (batch.length === 0) return; processBatch(batch); // 聚合计算 pendingCount -= batch.length; if (pendingCount > 0) { // 还有积压，放到下一个宏任务再处理 setTimeout(flushQueue, 0); } }这里的setTimeout(flushQueue, 0)是关键。它把一个大的同步任务切成多个小块，每块之间让出一帧给渲染。实测下来，对于 2000 条/秒的输入，这个策略能把单次脚本运行时间从 120ms 压到 20ms 以内。代价是报告延迟最多多出一个宏任务的时间，对于 1 秒窗口来说完全可接受。3.4 背压与丢弃策略：宁可丢，不可乱实时数据流最常见的问题就是“瞬间洪峰”：某台设备重连，一下补传过去 10 分钟的缓存数据，瞬间几千条冲进来。如果没有背压策略，聚合器会被淹没，内存飙升，窗口边界也被打乱。REA 的做法是给输入队列设上限bufferSize，超限后执行丢弃策略，并对外暴露一个drop事件，让调用方知道丢了多少数据。stream.on('drop', ({ count, oldestTimestamp }) => { console.warn(`已丢弃 ${count} 条事件，丢弃起始时间: ${oldestTimestamp}`); });丢弃策略我实现了两种：FIFO（丢最旧）和 LIFO（丢最新）。实时监控场景一般选 FIFO，因为旧数据对“当前状态”已经没有意义；但如果是做离线补传后的趋势分析，LIFO 可能更合适，保住最近的数据。默认是 FIFO，因为 REA 的首要目标是服务实时 UI，不是离线分析。3.5 去重：千万不能用 Set 去重整个事件dedupe这个配置一开始我想得很简单：用JSON.stringify(event)作为 key，存进 Set。压测立刻暴露问题：如果同一设备在同一毫秒上报了 2 条完全相同的温度值，确实会去重成功；但更多时候，两条事件只有timestamp不同，整体字符串完全不一样，Set 完全失效。而且真实场景里，相同设备相同值的重复上报才是常态，去重需求是“去掉连续的重复值”，不是“去掉所有相同的事件对象”。所以 REA 的去重设计成了可定制函数，默认提供一个defaultKey：function defaultKey(event: Record<string, unknown>): string { if (typeof event.deviceId === 'string' && typeof event.timestamp === 'number') { return `${event.deviceId}:${event.timestamp}:${event.type ?? 'event'}`; } return JSON.stringify(event); }同时支持调用方自己实现语义化去重，比如“同一设备 2 秒内的重复状态只保留一条”：const stream = createStream<DeviceEvent>({ dedupe: (event) => `${event.deviceId}:${Math.floor(event.timestamp / 2000)}:${event.status}` });4. 实测过程与性能记录4.1 测试环境与采集方法我用一个 Node.js 脚本压测 REA 的第一版稳定实现。没有用真实设备流，而是写了一个模拟器：每秒生产 2000 条设备事件，随机落到 16 台设备上，每台设备的温度和湿度在一定范围内波动。测试跑 5 分钟，采集三组数据：事件总数、内存峰值、脚本运行总耗时。对照组就是最朴素的“数组 push + setInterval 每秒渲染”的写法。# 模拟器伪代码 node bench/simulate.js --events-per-sec 2000 --duration 300 --window-ms 10004.2 关键指标对比方案事件总数（5min）内存峰值平均每帧脚本耗时报告输出量朴素数组60 万189 MB82ms60 万条原始数据REA v1（无分帧）60 万48 MB35ms约 3000 条报告REA v1（分帧）60 万51 MB18ms约 3000 条报告从表格能看出，REA 最显著的优势不是 CPU 占用而是“输出量”的缩减：60 万条事件被聚合为 3000 条报告，平均每个窗口输出 16 条左右，UI 层需要处理的节点数量少了两个数量级。内存方面，环形缓冲预分配让 GC 压力大幅降低，峰值稳定在 50MB 上下。4.3 第一版压测翻车的两个教训第一版实现里我用了一个“聪明”的方案：每次push的时候把新事件插入到数组的正确位置，让数组始终保持按时间戳有序。这个方案在 200 条/秒时运行得很好，但当事件超过 1000 条/秒时，二分查找加splice插入的开销迅速飙升，因为数组中间插入元素是 O(n)。压测到第 20 秒，单次push的耗时已经超过 3ms，事件积压瞬间崩塌。这个教训让我彻底放弃了“动态有序数组”，改用“环形缓冲 + 每次聚合前做分区”。环形缓冲本身就是按到达顺序排列的，聚合时再用双指针把“活窗口”拎出来。因为窗口数据总量是可控的（上限也就是 4000 条），线性遍历一次的代价是可以接受的。第二个教训是：一开始我没有做分帧，聚合计算放在push调用栈里同步执行。虽然计算本身很快，但 2000 条连续的同步push仍然会卡掉两到三帧。加上setTimeout(flushQueue, 0)分帧后，流畅度立刻提升了一个档次。4.4 时间边界：一个隐蔽的精度陷阱聚合报告里的时间戳字段，我最初直接拿Date.now()作为窗口边界，导致一个问题：多个客户端如果时钟不相同，同一窗口的起止时间会出现偏差。后来 REA 在配置里增加了timeSource，允许调用方自己指定时间来源，默认是() => Date.now()。这个细节看着小，但在多端联调时非常关键。如果前端图表和后端数据库用的时间基准不一致，你会在排查数据“为什么少了 1 秒”的时候浪费很多时间。5. 常见问题与排查清单5.1 使用 REA 时常见的 6 个问题问题现象根本原因解决方案窗口内明明有数据，但报告为空时间窗口按毫秒，事件里的timestamp是秒在push入口统一乘 1000 或配置timeUnit: 'ms'内存缓慢上涨，不下降分组 key 太多，每个分组的状态对象长期驻留给groupBy加白名单，或定期清理无事件分组的过期状态事件丢失但没日志队列入队失败，丢弃策略静默执行订阅drop事件，并配置告警阈值去重后数据仍重复默认去重 key 粒度太粗，没有包含关键字段自定义dedupe函数，按业务语义合成 key报告延迟不稳定事件突发导致分帧排队，报告输出被积压调小pollInterval到 50ms，或调大bufferSizeon('report')回调里操作 DOM 卡顿报告回调执行了重渲染在回调里再包一层requestAnimationFrame，配合虚拟列表5.2 一个必须自己动手做的实验我想特别强调上面表格里的第一行：时间精度。几乎每个接入 REA 的同事第一次联调都会踩到。最简单的排查方法是，在push入口打一行日志，打印event.timestamp和Date.now()，肉眼看一下数量级是否一致。如果event.timestamp是 10 位而Date.now()是 13 位，那不用查代码，问题就出在这里。5.3 自定义指标时容易忽略的计数条件metrics配置里的countIf需要传一个判定函数，有些同事会把判定函数写在配置外部，然后引用了外部变量。这在运行时会出问题：外部变量的值在窗口中途变化，统计口径就会变得不可预测。REA 内部要求判定函数必须是纯函数——同样的事件进来，永远返回同样的布尔值。如果你需要依赖外部状态，请把状态也编码进事件字段里，而不是直接闭包捕捉。结尾把 REA 从三个字母做成一个能跑、能测、能发布的小库，这个过程中我最大的体会是：实时数据处理的难点不在于算法有多么高深，而在于你愿不愿意为“不可能全量处理”这个前提做设计。环形缓冲不是什么新东西，增量聚合也是统计数据的基本功，但把它们组合成一个面向真实场景的库，需要反复在“简化 API”和“覆盖边界情况”之间取舍。如果你也要做类似的事件聚合工具，我的建议是：先写压测脚本，再写实现代码；先把drop事件设计进去，再谈优化；先把时间精度确认清楚，再连真实数据。这样至少能少熬几个夜。最后再分享一个小技巧：如果你不想依赖任何库，只想要一个“每秒输出一次最近窗口统计”的简易方案，完全可以用Map加setInterval实现，数据结构就两样——一个按时间戳排序的数组，一个分组统计对象。REA 的本意也不是让你一定用它，而是希望这套思路能帮你设计出自己的“数据降噪层”。
半年前我在本地新建了一个空目录，名字就叫rea，没有说明文档，没有 package.json，只有三个字母。后来我给它补全了一个含义：REA = Real-time Event Aggregator，一个轻量级实时事件聚合器。这算是把项目名从“名字”变成“定义”的过程。如果你正在做监控大屏、设备遥测、实时日志这类前端页面，大概率会遇到同一个问题：后端推送的数据太密、太碎，直接渲染会卡，全量存内存会爆，而业务上你真正需要的，可能只是“每秒动态、每10秒均值、异常数”这些东西。REA 就是用来做这层加工的工具。这篇文章我按项目复盘的思路来写：为什么需要它、API 是怎么设计的、核心实现里有哪些细节和坑、压测表现怎么样。里面涉及的所有代码都是我在实际开发中写过的简化版，可以直接拿去改。1. 为什么需要 REA：一个实时数据小场景1.1 你遇到过的“数据风暴”之前我在做一个设备状态面板，后端通过 MQTT 网关推送传感器数据，频率大概每秒 2000 条。每条数据长这样：{ "deviceId": "dev-001", "timestamp": 1735123456789, "temperature": 23.4, "humidity": 43.2, "status": "online" }刚开始我图省事，前端直接在onmessage里把数据push进一个数组，然后用setInterval每秒取一次数组更新图表。结果启动不到两分钟，页面就开始掉帧，DevTools 里内存曲线一路爬升到 200MB。原因很直接：5 万条数据全部堆积在内存里，图表库每次渲染都要遍历整个数组，大量 DOM 更新被阻塞。更要命的是，这种“实时数据”里超过 80% 是冗余的——同一台设备一秒钟上报 5 次温度，屏幕上的折线图根本看不出来差异。我在那一瞬间意识到，前端需要的不只是“收到就显示”，而是“把收到的东西加工成有意义的信号”之后再显示。这个加工层，就是实时事件聚合。1.2 为什么不在后端做聚合有人会问：后端直接聚合好再推给前端不行吗？可以，但现实往往不允许。后端的推送通道通常是通用数据管道，面向多个消费方——有人要原始数据做离线分析，有人要按设备聚合，有人要全量日志，让后端为每一种前端需求单独设计聚合接口，成本和沟通周期都很高。你要的是一个“滑动窗口内求均值”的口径，改一次需求就要后端改一次推送结构，联调一遍，这在快速迭代的场景里非常拖节奏。所以我把聚合逻辑放在前端，做成一个独立库，由调用方自己决定窗口大小、分组维度、统计指标。这也是 REA 核心的设计立场：原始数据不动，分析口径由消费方定义。1.3 REA 与防抖节流的本质区别很多同事第一次看到 REA 会问“这不就是 lodash 的 throttle 吗？”还真不是。防抖和节流解决的是“调用频率”问题，它们只关心一个动作多久执行一次；而聚合解决的是“数据到信息”的转换问题，它要做时间窗口、分组、去重、统计和采样。举一个例子：一批设备状态消息，1 秒内来了 2000 条。节流能做到“这 2000 条只触发一次 UI 更新”，但 UI 拿到的还是 2000 条原始数据，该卡还是卡。REA 能把这 2000 条变成“16 台设备各自的平均温度、最高湿度、在线率”，输出一个体积很小的报告。用摄影类比：防抖节流是让你别连续按快门，聚合是先把拍到的几百张照片自动筛选拼接成一张全景图，你拿到的是结果，不是底片。2. REA 的 API 设计与核心模型2.1 三句话描述 REAREA 对外暴露一个核心方法createStream，它返回一个流对象。你通过push把原始事件投进去，通过on订阅聚合后的报告，中间用window、groupBy、metrics配置来决定“怎么聚”。整个过程是一个链式管道。第一版我把它设计成纯粹的面向对象风格，每个统计器是一个类。后来发现类层次太深，调用方自定义指标时很别扭。重构后改成函数式管道：一个流就是一组小模块的串联，最后接一个订阅回调。这样既好测试，也好扩展。import { createStream } from 'rea'; interface DeviceEvent { deviceId: string; timestamp: number; temperature: number; humidity: number; status: 'online' | 'offline'; } const stream = createStream<DeviceEvent>({ aggregate: { // 每 1000ms 形成一个窗口 windowMs: 1000, // 按设备分组 groupBy: 'deviceId', metrics: { avgTemp: ['temperature', 'avg'], maxHum: ['humidity', 'max'], offlineCount: ['status', 'countIf', v => v === 'offline'] } } }); stream.on('report', (report) => { renderDeviceCards(report.groups); }); // 模拟消息入口 stream.push({ deviceId: 'dev-001', timestamp: Date.now(), temperature: 23.5, humidity: 41, status: 'online' });2.2 为什么不用现成的流式库在 REA 之前，我考虑过直接引入响应式编程库。它们功能确实完整，冷热流、背压、调度器、操作符一应俱全。但引入到项目里之后我发现自己只需要其中三个操作：按时间切片、按字段分组、算几个指标。而完整库带来的概念负担，比如 Subject、Observable、Scheduler，对于只想要一个“数据加工器”的团队来说太重了。REA 刻意缩小了范围：没有高阶流、没有异步调度编排、没有操作符组合子。流对象本质上是“一个可推送的队列加一组归约函数”，你给它数据，它按配置吐报告。世界上的问题不是都要用图灵完备的抽象来解决，一个跑步机比一台跑车更适合放在客厅里。2.3 核心 API 与配置项用表格整理一下第一版稳定 API，方便你对照自己的场景。API / 配置类型说明createStream(options)函数创建聚合流，返回 Stream 实例stream.push(event)方法推入一条原始事件stream.on('report', cb)方法订阅聚合报告，窗口结束时触发stream.start()/stream.stop()方法手动启停内部定时器windowMsnumber统计窗口时长，单位毫秒groupBystring | (event) => string分组字段或自定义分组函数metricsRecord<string, Metric>定义输出指标，metrics 内部是归约配置emitEmptyboolean窗口内无数据时是否输出空报告，默认 falsededupeboolean | (event) => string去重开关或自定义去重键函数pollIntervalnumber检查窗口边界的时间间隔，默认 100msbufferSizenumber输入缓冲队列上限，超过后按丢弃策略处理第一版没有配置项乱飞的毛病，是在使用中慢慢加出来的：emitEmpty是监控大屏需要水位线，即使没有新数据也要让 UI 保持“活着”的视觉效果；bufferSize是压测后加的，事件爆发时宁可丢不要爆。2.4 用类型把事件结构约束住实时事件数据的结构，有时候真的只有上线那一刻才知道。所以 REA 在 TypeScript 泛型上花了一些功夫：DeviceEvent这个 interface 可以完全由使用者定义，groupBy的类型会根据传入的 interface 自动推导，metrics里指定的字段如果不存在，编译期就会报错。// 错误示例：编译不通过 const stream = createStream<DeviceEvent>({ aggregate: { metrics: { invalidMetric: ['nonExistentField', 'avg'] // 报错：DeviceEvent 没有 nonExistentField } } });这算是我自己比较满意的一个设计点：让“配置写错”从运行时的静默异常，变成编译期的红波浪线。团队新人接手配置时，不容易把字段名拼错。3. 核心实现细节与调优3.1 滑动时间窗口：不是定时清数组第一时间想到的实现方案肯定是：开一个setInterval，每 1 秒把数组里的东西全算一遍，然后清空数组。这个方案的问题在于时间边界——如果一条事件的timestamp是 00:00:00.950，另一条是 00:00:01.050，按照绝对秒来切窗口，它们会被分到两个窗口。但对于一个“最近 1 秒”的滑动窗口，它们其实应该落在同一个窗口里。要正确实现滑动窗口，内存里不应该存“一个窗口的数据”，而应该存“一个时间范围内的事件”。我用的是环形缓冲加“过期指针”的方案：固定长度数组存储事件，头部指针和尾部指针分别指向最早和最晚事件；每次push时，把事件写到尾部，尾部指针后移；每次触发聚合时，从头部开始向前遍历，把timestamp超出windowMs的事件标记为过期，头部指针后移；剩下的数据就是“活窗口”。这样做的好处是，聚合操作不需要反复filter全量数组，遍历的永远只是活数据。我用的环形缓冲简化版大概是这样的：class RingBuffer<T extends { timestamp: number }> { private buffer: (T | null)[]; private head = 0; private tail = 0; private size = 0; constructor(private capacity: number) { this.buffer = new Array(capacity).fill(null); } push(item: T): boolean { if (this.size === this.capacity) { return false; // 队列满，丢弃策略由外部决定 } this.buffer[this.tail] = item; this.tail = (this.tail + 1) % this.capacity; this.size++; return true; } expireBefore(timestamp: number): void { while (this.size > 0) { const item = this.buffer[this.head]; if (!item || item.timestamp >= timestamp) break; this.buffer[this.head] = null; this.head = (this.head + 1) % this.capacity; this.size--; } } *activeItems(): Generator<T> { for (let i = 0; i < this.size; i++) { const idx = (this.head + i) % this.capacity; const item = this.buffer[idx]; if (item) yield item; } } }这个方案有两个真实好处：一是内存占用是预分配的，不会因为数据量上涨导致频繁 GC；二是“过期清理”是 O(1) 操作，不是每次聚合并遍历全量。初始容量根据业务峰值定，比如capacity = maxEventsPerSecond * 2，不够再扩容。注意：环形缓冲里的capacity要留出余量。如果每秒 2000 条、窗口 1 秒，缓冲容量至少开 4000，给事件突发和窗口重叠留空间。3.2 聚合统计：增量计算替代重复遍历窗口内 2000 条事件，如果每次报告都把所有事件重新求和再求平均，代价并不大——2000 次加法对 CPU 来说毫秒级。但当窗口变成 5 秒、事件变成每秒 1 万条，这种朴素算法的性能就不够看了。REA 在指标计算上做了两层优化：第一层，增量归约。每次push事件时，不是等窗口结束才计算，而是在事件进来时立刻更新一个聚合状态：累计值、最大值、最小值、计数。窗口结束时只需读取这个状态，O(1) 时间就能给出报告。为了让“求平均”这类指标能被增量计算而不存储原始数据，我定义了AggregateState接口：interface AggregateState { sum: number; count: number; max: number; min: number; latest: number; }每来一条事件，sum += value; count++; if (value > max) max = value，这样窗口结束直接sum / count就是平均。内存占用从“存 N 条事件”降到“存 N 个状态”。第二层，事件类型分桶。REA 内部把事件先按groupBy分组，每个分组独立维护聚合状态。这样同一个设备的事件只会更新属于自己的那一个状态，不会出现所有设备混在一个计数器里的口径错误。3.3 分帧处理：把 2000 条推送掰成几次小任务即使聚合计算再快，如果push是在同一个事件循环 tick 里连续处理 2000 次，UI 线程仍然会被长时间占用。这里的核心问题是：数据进入的速度不应该完全决定处理的节奏。REA 采用了一种“分帧批处理”的策略：push只做入队动作，复杂度 O(1)；聚合计算由一个调度器触发，默认每pollInterval（100ms）处理一次队列里的所有事件，但如果一次要处理的事件超过阈值（比如 500 条），就把这批事件拆到多个宏任务里处理，让浏览器有时间插队渲染。// 伪代码：分帧调度 let pendingCount = 0; const MAX_BATCH = 500; function flushQueue() { const batch = queue.dequeueUpTo(MAX_BATCH); if (batch.length === 0) return; processBatch(batch); // 聚合计算 pendingCount -= batch.length; if (pendingCount > 0) { // 还有积压，放到下一个宏任务再处理 setTimeout(flushQueue, 0); } }这里的setTimeout(flushQueue, 0)是关键。它把一个大的同步任务切成多个小块，每块之间让出一帧给渲染。实测下来，对于 2000 条/秒的输入，这个策略能把单次脚本运行时间从 120ms 压到 20ms 以内。代价是报告延迟最多多出一个宏任务的时间，对于 1 秒窗口来说完全可接受。3.4 背压与丢弃策略：宁可丢，不可乱实时数据流最常见的问题就是“瞬间洪峰”：某台设备重连，一下补传过去 10 分钟的缓存数据，瞬间几千条冲进来。如果没有背压策略，聚合器会被淹没，内存飙升，窗口边界也被打乱。REA 的做法是给输入队列设上限bufferSize，超限后执行丢弃策略，并对外暴露一个drop事件，让调用方知道丢了多少数据。stream.on('drop', ({ count, oldestTimestamp }) => { console.warn(`已丢弃 ${count} 条事件，丢弃起始时间: ${oldestTimestamp}`); });丢弃策略我实现了两种：FIFO（丢最旧）和 LIFO（丢最新）。实时监控场景一般选 FIFO，因为旧数据对“当前状态”已经没有意义；但如果是做离线补传后的趋势分析，LIFO 可能更合适，保住最近的数据。默认是 FIFO，因为 REA 的首要目标是服务实时 UI，不是离线分析。3.5 去重：千万不能用 Set 去重整个事件dedupe这个配置一开始我想得很简单：用JSON.stringify(event)作为 key，存进 Set。压测立刻暴露问题：如果同一设备在同一毫秒上报了 2 条完全相同的温度值，确实会去重成功；但更多时候，两条事件只有timestamp不同，整体字符串完全不一样，Set 完全失效。而且真实场景里，相同设备相同值的重复上报才是常态，去重需求是“去掉连续的重复值”，不是“去掉所有相同的事件对象”。所以 REA 的去重设计成了可定制函数，默认提供一个defaultKey：function defaultKey(event: Record<string, unknown>): string { if (typeof event.deviceId === 'string' && typeof event.timestamp === 'number') { return `${event.deviceId}:${event.timestamp}:${event.type ?? 'event'}`; } return JSON.stringify(event); }同时支持调用方自己实现语义化去重，比如“同一设备 2 秒内的重复状态只保留一条”：const stream = createStream<DeviceEvent>({ dedupe: (event) => `${event.deviceId}:${Math.floor(event.timestamp / 2000)}:${event.status}` });4. 实测过程与性能记录4.1 测试环境与采集方法我用一个 Node.js 脚本压测 REA 的第一版稳定实现。没有用真实设备流，而是写了一个模拟器：每秒生产 2000 条设备事件，随机落到 16 台设备上，每台设备的温度和湿度在一定范围内波动。测试跑 5 分钟，采集三组数据：事件总数、内存峰值、脚本运行总耗时。对照组就是最朴素的“数组 push + setInterval 每秒渲染”的写法。# 模拟器伪代码 node bench/simulate.js --events-per-sec 2000 --duration 300 --window-ms 10004.2 关键指标对比方案事件总数（5min）内存峰值平均每帧脚本耗时报告输出量朴素数组60 万189 MB82ms60 万条原始数据REA v1（无分帧）60 万48 MB35ms约 3000 条报告REA v1（分帧）60 万51 MB18ms约 3000 条报告从表格能看出，REA 最显著的优势不是 CPU 占用而是“输出量”的缩减：60 万条事件被聚合为 3000 条报告，平均每个窗口输出 16 条左右，UI 层需要处理的节点数量少了两个数量级。内存方面，环形缓冲预分配让 GC 压力大幅降低，峰值稳定在 50MB 上下。4.3 第一版压测翻车的两个教训第一版实现里我用了一个“聪明”的方案：每次push的时候把新事件插入到数组的正确位置，让数组始终保持按时间戳有序。这个方案在 200 条/秒时运行得很好，但当事件超过 1000 条/秒时，二分查找加splice插入的开销迅速飙升，因为数组中间插入元素是 O(n)。压测到第 20 秒，单次push的耗时已经超过 3ms，事件积压瞬间崩塌。这个教训让我彻底放弃了“动态有序数组”，改用“环形缓冲 + 每次聚合前做分区”。环形缓冲本身就是按到达顺序排列的，聚合时再用双指针把“活窗口”拎出来。因为窗口数据总量是可控的（上限也就是 4000 条），线性遍历一次的代价是可以接受的。第二个教训是：一开始我没有做分帧，聚合计算放在push调用栈里同步执行。虽然计算本身很快，但 2000 条连续的同步push仍然会卡掉两到三帧。加上setTimeout(flushQueue, 0)分帧后，流畅度立刻提升了一个档次。4.4 时间边界：一个隐蔽的精度陷阱聚合报告里的时间戳字段，我最初直接拿Date.now()作为窗口边界，导致一个问题：多个客户端如果时钟不相同，同一窗口的起止时间会出现偏差。后来 REA 在配置里增加了timeSource，允许调用方自己指定时间来源，默认是() => Date.now()。这个细节看着小，但在多端联调时非常关键。如果前端图表和后端数据库用的时间基准不一致，你会在排查数据“为什么少了 1 秒”的时候浪费很多时间。5. 常见问题与排查清单5.1 使用 REA 时常见的 6 个问题问题现象根本原因解决方案窗口内明明有数据，但报告为空时间窗口按毫秒，事件里的timestamp是秒在push入口统一乘 1000 或配置timeUnit: 'ms'内存缓慢上涨，不下降分组 key 太多，每个分组的状态对象长期驻留给groupBy加白名单，或定期清理无事件分组的过期状态事件丢失但没日志队列入队失败，丢弃策略静默执行订阅drop事件，并配置告警阈值去重后数据仍重复默认去重 key 粒度太粗，没有包含关键字段自定义dedupe函数，按业务语义合成 key报告延迟不稳定事件突发导致分帧排队，报告输出被积压调小pollInterval到 50ms，或调大bufferSizeon('report')回调里操作 DOM 卡顿报告回调执行了重渲染在回调里再包一层requestAnimationFrame，配合虚拟列表5.2 一个必须自己动手做的实验我想特别强调上面表格里的第一行：时间精度。几乎每个接入 REA 的同事第一次联调都会踩到。最简单的排查方法是，在push入口打一行日志，打印event.timestamp和Date.now()，肉眼看一下数量级是否一致。如果event.timestamp是 10 位而Date.now()是 13 位，那不用查代码，问题就出在这里。5.3 自定义指标时容易忽略的计数条件metrics配置里的countIf需要传一个判定函数，有些同事会把判定函数写在配置外部，然后引用了外部变量。这在运行时会出问题：外部变量的值在窗口中途变化，统计口径就会变得不可预测。REA 内部要求判定函数必须是纯函数——同样的事件进来，永远返回同样的布尔值。如果你需要依赖外部状态，请把状态也编码进事件字段里，而不是直接闭包捕捉。结尾把 REA 从三个字母做成一个能跑、能测、能发布的小库，这个过程中我最大的体会是：实时数据处理的难点不在于算法有多么高深，而在于你愿不愿意为“不可能全量处理”这个前提做设计。环形缓冲不是什么新东西，增量聚合也是统计数据的基本功，但把它们组合成一个面向真实场景的库，需要反复在“简化 API”和“覆盖边界情况”之间取舍。如果你也要做类似的事件聚合工具，我的建议是：先写压测脚本，再写实现代码；先把drop事件设计进去，再谈优化；先把时间精度确认清楚，再连真实数据。这样至少能少熬几个夜。最后再分享一个小技巧：如果你不想依赖任何库，只想要一个“每秒输出一次最近窗口统计”的简易方案，完全可以用Map加setInterval实现，数据结构就两样——一个按时间戳排序的数组，一个分组统计对象。REA 的本意也不是让你一定用它，而是希望这套思路能帮你设计出自己的“数据降噪层”。