At the edge, the number that matters is memory - not throughput (specially in Ramageddon)

在边缘计算中，最重要的指标是内存，而非吞吐量（尤其是在“内存灾难”时期）

We rebuilt LF Edge eKuiper in Rust and ran it against eKuiper, Telegraf and Redpanda Connect on five real MQTT workloads — one core, 1 GB of memory, and output checked message-by-message. The most important result wasn’t speed. 我们使用 Rust 重构了 LF Edge eKuiper，并将其与 eKuiper、Telegraf 和 Redpanda Connect 在五个真实的 MQTT 工作负载上进行了对比测试——环境配置为单核 CPU、1 GB 内存，并对输出结果进行了逐条校验。测试结果中最重要的一点并非速度。

Most stream-processing benchmarks you’ll read optimize for one number: peak throughput on a big server. That number is close to useless for the place these engines actually run — an industrial gateway, an ESPHome hub, a vehicle head-unit, an EV charger. There, you get one or two CPU cores and a few hundred megabytes of free memory, your input arrives over MQTT, and your traffic is bursty in the worst way: fleets reconnect together, chargers start sessions together, devices flush buffered readings all at once after an outage. 你所读到的大多数流处理基准测试都只优化一个指标：大型服务器上的峰值吞吐量。但对于这些引擎实际运行的场景——工业网关、ESPHome 集线器、车载主机、电动汽车充电桩——这个数字几乎毫无意义。在这些场景中，你通常只有一两个 CPU 核心和几百兆的可用内存，输入数据通过 MQTT 到达，且流量呈现出最糟糕的突发性：整个车队同时重连、充电桩同时开启会话、设备在断网恢复后同时刷新缓存的读数。

In that world two questions decide whether your pipeline survives, and neither is peak throughput: Does the engine keep up on a single core? Does its memory stay bounded when traffic grows? We built a stream engine called rekuiper to answer “yes” to both, and then we built a benchmark honest enough to tell us if we’d actually managed it. 在那个世界里，决定你的流水线能否存活的只有两个问题，且都不是峰值吞吐量：引擎在单核上能否跟上处理速度？当流量增长时，内存占用是否保持在受控范围内？我们构建了一个名为 rekuiper 的流引擎，旨在对这两个问题给出肯定的回答，随后我们又构建了一个足够客观的基准测试，以验证我们是否真正实现了目标。

This post is about the benchmark as much as the engine, because the benchmark taught us more than we expected — including a correctness bug in our own code that a throughput-only test would have rewarded as “fast.” The headline: across five MQTT workloads shaped like real deployments, rekuiper produced complete, correct output at 100,000 messages per second on one core in every workload — the top of our tested range, so we never found its ceiling. 这篇文章既是关于引擎的，也是关于基准测试的，因为测试过程教会了我们比预期更多的东西——包括我们代码中一个正确性 Bug，如果只进行吞吐量测试，它反而会被标记为“快速”。核心结论是：在五个模拟真实部署的 MQTT 工作负载中，rekuiper 在单核上均能以每秒 10 万条消息的速度产生完整、正确的输出——这已达到我们测试范围的上限，因此我们从未触及其性能瓶颈。

But the result we care about most isn’t that. It’s that on the windowed workloads, rekuiper’s memory stayed between 5 and 10 MB while the Go-based engines climbed to half a gigabyte to a full gigabyte, or failed. That gap is the whole point, and it comes from design, not from the language. 但我们最关心的结果并非如此。在窗口化工作负载中，rekuiper 的内存占用保持在 5 到 10 MB 之间，而基于 Go 语言的引擎则攀升至 0.5 GB 到 1 GB，甚至直接崩溃。这种差距才是关键所在，它源于设计，而非编程语言本身。

What rekuiper is

什么是 rekuiper

rekuiper is a stream-processing engine written in Rust that reimplements the surface of LF Edge eKuiper: its REST API, its SQL dialect, its stream and rule definitions, and its kuiper command-line interface. The goal was boring on purpose — existing eKuiper rules, the eKuiper Manager web UI, and deployment tooling should keep working — so that “switch the engine” isn’t also “rewrite everything.” rekuiper 是一个用 Rust 编写的流处理引擎，它重新实现了 LF Edge eKuiper 的接口层：包括 REST API、SQL 方言、流与规则定义，以及 kuiper 命令行界面。我们的目标是有意保持“平庸”——现有的 eKuiper 规则、eKuiper Manager Web UI 和部署工具都应能继续工作——这样“切换引擎”就不意味着“重写一切”。

Concretely, the compatibility surface covers eKuiper’s REST API (98 paths and 140 operations, checked black-box against eKuiper’s own OpenAPI description), the SQL dialect including JSON paths, CASE, array indexing and unnest, and eKuiper’s stream option names (DATASOURCE, FORMAT, CONF_KEY, SCHEMAID, TIMESTAMP, and so on). If you know eKuiper, you already know rekuiper. What’s different is underneath, and it’s built around one principle: memory stays bounded under load. Three design choices carry that, and each one shows up later in the numbers. 具体来说，兼容性覆盖了 eKuiper 的 REST API（98 个路径和 140 个操作，通过 eKuiper 自身的 OpenAPI 描述进行了黑盒校验）、SQL 方言（包括 JSON 路径、CASE、数组索引和 unnest），以及 eKuiper 的流选项名称（DATASOURCE、FORMAT、CONF_KEY、SCHEMAID、TIMESTAMP 等）。如果你了解 eKuiper，你就已经了解 rekuiper。不同之处在于底层，它围绕一个原则构建：在高负载下内存保持受控。三个设计选择支撑了这一点，每一个都在后续的数据中得到了体现。

Three design choices that keep memory flat

三个保持内存平稳的设计选择

Bounded queues with real backpressure 带有真实背压的有限队列

Sources publish records into an in-process stream bus with bounded per-subscriber queues — 4,096 records each. Admission is reserve-then-commit: a batch first reserves capacity in every subscriber’s queue, and only then commits. So a batch is either delivered to all subscribers or rejected outright, never half-delivered, and a slow rule pushes back on its source instead of quietly dropping data. 数据源将记录发布到进程内的流总线中，每个订阅者拥有有限的队列（每个 4,096 条记录）。准入机制采用“预留-提交”模式：批处理首先在每个订阅者的队列中预留容量，然后才提交。因此，一个批次要么被交付给所有订阅者，要么被直接拒绝，绝不会出现交付一半的情况。如果某条规则处理缓慢，它会向数据源施加背压，而不是悄悄丢弃数据。

Each rule runs as its own task, and its output drains through a bounded sink queue (default 10,000) served by a dedicated sink worker. The MQTT source uses the rumqttc client, and when a single network read surfaces several publishes, the source admits everything already buffered as one batch of up to 1,024 records. That avoids a per-message wakeup without ever waiting around for more data — you pay one scheduling cost for a burst instead of one per message. This batch-admission trick is a big part of why rekuiper uses roughly half the CPU per message of the Go engines on the simple workloads. 每条规则作为独立的任务运行，其输出通过一个有限的接收队列（默认 10,000）排出，并由专门的接收工作线程处理。MQTT 数据源使用 rumqttc 客户端，当单次网络读取出现多个发布消息时，数据源会将所有已缓冲的内容作为一个批次（最多 1,024 条记录）进行准入。这避免了每条消息唤醒一次的开销，且无需等待更多数据——你只需为一次突发流量支付一次调度成本，而不是每条消息支付一次。这种批处理准入技巧是 rekuiper 在简单工作负载下每条消息 CPU 开销仅为 Go 引擎一半的主要原因。

Incremental window aggregation: O(groups), not O(messages) 增量窗口聚合：O(groups)，而非 O(messages)

This is the important one. When you compute GROUP BY device, TUMBLINGWINDOW(ss, 10) with count, avg, max and friends, the naive way is to buffer every row that falls in the window and aggregate at the trigger. Memory then grows with traffic — messages per window — which is exactly the thing that explodes when a fleet reconnects. 这是最重要的一点。当你计算 GROUP BY device、TUMBLINGWINDOW(ss, 10) 并配合 count、avg、max 等函数时，原始的做法是缓冲窗口内的每一行数据，并在触发时进行聚合。此时内存会随流量（每个窗口的消息数）增长——这正是车队重连时导致内存爆炸的原因。

rekuiper instead keeps one accumulator per group per aggregate and never stores the rows. Window memory becomes a function of the number of devices, not the number of messages. For the common edge shape — group columns, plain columns, and count/sum/avg/min/max over simple expressions — this incremental evaluator does the whole job. Statements that genuinely need the rows (collect(), joins, some HAVING) fall back to a buffered evaluator, and a unit test checks the two produce identical output on mixed data. For a fleet, this is the difference between memory that scales with how many vehicles you have and memory that scales with how fast they’re all talking at once. Only one of those is safe on a 1 GB box. 相反，rekuiper 为每个组的每个聚合维护一个累加器，从不存储原始行。窗口内存变成了设备数量的函数，而不是消息数量的函数。对于常见的边缘计算场景——分组列、普通列以及对简单表达式进行 count/sum/avg/min/max——这种增量评估器可以完成所有工作。对于确实需要原始行的语句（如 collect()、joins、部分 HAVING），则回退到缓冲评估器，单元测试会确保两者在混合数据上产生相同的输出。对于车队而言，这意味着内存占用是随车辆数量扩展，还是随车辆通信频率扩展。在 1 GB 内存的设备上，只有前者是安全的。

An offline sink cache that spills instead of dropping 离线接收缓存：溢出而非丢弃

For intermittent uplinks — a vehicle in a tunnel, a remote site on flaky cellular — a sink can enable a cache using eKuiper’s own options (enableCache, memoryCacheThreshold, maxDiskCache, and the rest). Records whose send fails recoverably are queued FIFO: in memory up to a threshold, then in disk pages, and only when the disk budget is exhausted are the oldest records dropped — and counted, not silently lost. The MQTT sink holds one persistent connection per action and reports disconnection, so an outage is detected and cached rather than quietly discarded. (The cache is covered by an integration test but isn’t part of the performance numbers here.) 对于间歇性上行链路场景——如隧道中的车辆、蜂窝网络不稳定的远程站点——接收端可以使用 eKuiper 原有的选项（enableCache、memoryCacheThreshold、maxDiskCache 等）启用缓存。发送失败且可恢复的记录将按 FIFO（先进先出）排队：先在内存中缓存直到阈值，然后溢出到磁盘页面，只有当磁盘配额耗尽时，最旧的记录才会被丢弃——且会被计数，而不是悄无声息地丢失。MQTT 接收端为每个动作保持一个持久连接并报告断开，因此中断会被检测到并缓存，而不是被静默丢弃。（缓存功能包含在集成测试中，但未计入此处的性能数据。）

The benchmark that doesn’t lie to you 不会对你撒谎的基准测试

Here’s the uncomfortable truth about a lot of edge stream-processing comparisons: they measure throughput at the point the engine acknowledges ingest, or they count output records without checking that the records a… 关于许多边缘流处理对比测试，有一个令人不安的事实：它们要么在引擎确认接收时测量吞吐量，要么在不校验记录内容的情况下统计输出记录数量……