从批处理到实时：轻量级数据管道rea的设计与落地实践

前阵子有个朋友问我：你们服务端的访问量统计，为什么非要从每天凌晨跑批改成实时管道？原来那种定时生成报表的方式，不也用得好好的吗？这个问题让我认真想了很久。答案其实很直接——当业务人员开始盯着"过去五分钟的数据"做决策时，批处理方案就已经走到了尽头。为了彻底回答他，也为了验证我对整套实时数据链路的设计想法，我动手写了 rea 这个项目。

rea 是我最近正在推进的一个轻量级实时数据管道项目，定位介于"全量上大数据组件"和"简单日志采集"之间，目标是让一个普通后端团队用尽量少的组件，把分散在各服务里的业务事件在秒级内汇聚到统一存储，并支撑报表查询与简单告警。这个项目覆盖数据采集、缓冲队列、消费加工、落库存储、查询展示五个环节，适合正在从批处理向实时链路过渡、但又不想一上来就引入重型分布式框架的团队参考。

1. 为什么我要自己造一个 rea：批处理方案的真实短板

1.1 批处理让我抓狂的三个具体场景

先说第一个让我下决心的场景：每天凌晨的统计任务。我维护过一套按小时分片的批处理任务，每小时从各业务库拉一遍数据，汇总后写入结果表。表面看没什么问题，但一旦某张源表的数据量涨上去，任务执行时间就从十分钟变成四十分钟，再变成凌晨三点还在跑。数据部门第二天早上九点看到的报表，其实是昨晚十一点之前的快照，中间整整九个小时都是盲区。

第二个场景更让人头疼：故障定位。有一次线上接口的异常率突然升高，但异常日志要等批处理任务跑完才能进入统计库，等我拿到聚合结果时，已经过去了两个多小时。那段时间我只能靠临时翻日志、手动数错误码来排查，效率低到离谱。事后我在想，如果有一条秒级可见的管道，异常率的变化曲线当场就能拉出来，定位时间至少能缩短一个数量级。

第三个场景是关于业务侧的反馈。业务负责人不止一次问过：能不能在运营后台直接看到当前的订单量、注册量？我说只能看昨天的，然后就没有然后了。做技术的都明白，这不是加个接口就能解决的问题，是整个数据流转方式的差异。也就是从那时起，我意识到"实时"不是一个可有可无的优化项，而是业务发展到了某个阶段后的必然要求。

1.2 对比现成方案后，rea 选择轻量自研的理由

决定做实时链路时，我其实先考虑过直接用现成的重型组件：分布式消息队列加流处理框架，再加一套列式存储。这套组合确实是业界标准方案，处理海量数据时表现很稳定，但对于一个日吞吐量几千万事件、查询并发也不高的团队来说，带来的运维负担明显大于收益。要维护的组件数量多了一倍，出问题时排查链路也长，人员成本一下子就上去了。

我把三种候选路线的核心差异整理了一下，最终选择了自研轻量管道这条中间路线：

表格里的结论很清晰。rea 的目标不是做大而全的平台，而是用最少组件覆盖"采集、缓冲、消费、存储"这条主干。组件少，意味着故障域小，任何一个节点出问题，都能快速定位到具体模块。对于小团队来说，这种可控性比性能上限更重要。

补充一点：我这里说的自研，并不是从零重写消息队列，而是把现有组件以最朴素的方式组合起来，自己只写采集探针、消费逻辑和存储模型。真正需要创造代码的部分其实很少，大部分工作是在设计数据契约和确认数据口径。

2. rea 的骨架设计：一条轻量链路如何串联采集、缓冲与存储

2.1 rea 的整体链路：采集、缓冲、加工、存储、展示

rea 的整体数据流向是：各业务服务通过 SDK 将事件发送到统一接入点，接入点将原始数据写入缓冲队列，消费服务从队列读取数据后做清洗和指标计算，再批量写入存储层，最后由查询接口对上层提供聚合查询能力。

这五个环节里，我重点讲三个设计决策。

第一，接入点采用独立部署的轻量 HTTP 服务，而不是让业务服务直连队列组件。原因很实际：队列组件的连接数有限，如果几百个服务实例同时建连，队列端压力会很大；而 HTTP 接入点可以横向扩容，前边加一层负载均衡就能扛住流量波动。同时，接入点和缓冲队列解耦后，队列组件升级或迁移时，业务侧完全无感。

第二，缓冲层用 Redis Stream 来承担，而不是 Redis List。Stream 是 Redis 5.0 引入的数据结构，天然支持消费者组、消息确认和待处理列表，这三点正好是构建可靠消费链路缺一不可的。消费者组允许多个消费实例分摊数据，消息确认能防止消费中途崩溃丢数据，待处理列表方便运维查看积压了多少消息。List 虽然实现简单，但要做消费位点记录，得自己额外维护一套逻辑，得不偿失。

第三，存储层选择关系型数据库加分区表，而不是时序数据库。rea 处理的事件带有明确的时间戳，最常用的查询是"某分钟内某个接口的访问量是多少"，分区表按天分片后，扫表范围可以控制在一个分区内，性能完全够用。时序数据库在压缩率和查询能力上确实更强，但多一个组件就多一份运维成本，在数据量还没到那个量级之前，我不想过早引入。

2.2 数据契约先行：定义事件体与各模块接口

写代码之前，我花了整整两天定义数据契约。这是我觉得 rea 做对的一件重要事情。如果没有统一的事件格式，后续每个环节都可能出现字段名不统一、类型不一致的问题，越往后返工成本越高。

事件体最终定成了这样一份 JSON 结构：

{ "event_id": "uuid", "event_type": "order_created", "event_time": "2023-06-15T14:30:01+08:00", "service": "trade-service", "instance": "192.168.1.10:8080", "dimensions": { "user_id": "u_10086", "channel": "ios" }, "metrics": { "order_amount": 99.8, "item_count": 2 } }

设计这份结构时遵循了几个原则：event_id 全链路唯一，用来做去重；event_type 用于区分业务场景，后续消费端做指标计算时直接按这个字段分流；dimensions 和 metrics 分离，前者是所有维度字段的容器，后者是数值型指标，这样后端做聚合时可以直接取 metrics 里的字段套用聚合函数。

模块之间的接口我也做了明确约定。采集探针到接入点之间是 HTTP JSON 上报接口，接入点到缓冲队列之间是 Redis Stream 消息写入，消费服务到存储层之间是批量 SQL 写入。每个接口的语义都在代码仓库里维护了一份协议文档，改任何字段都要先改文档再改代码。这一步看起来繁琐，但在后来实际开发中帮我避免了很多无效沟通。

3. 实现细节：从采集探针到幂等消费的核心代码

3.1 采集端实现：批量上报与降级策略

采集探针是嵌入各业务服务的客户端，它要解决的核心问题是"如何尽量不影响业务线程的响应时间"。如果每产生一条事件就同步发一次 HTTP 请求，业务接口的延迟会肉眼可见地增加，这是谁都没法接受的。所以探针内部必须有一个缓冲队列，积攒到一定数量或一定时间后再批量上报。

我用 Python 写的简易版本核心逻辑如下：

import threading import time import requests class ReaProbe: BATCH_SIZE = 100 FLUSH_INTERVAL = 2 def __init__(self, endpoint): self.endpoint = endpoint self._buffer = [] self._lock = threading.Lock() self._timer = threading.Thread(target=self._periodic_flush, daemon=True) self._timer.start() def emit(self, event): with self._lock: self._buffer.append(event) if len(self._buffer) >= self.BATCH_SIZE: self._flush_unlocked() def _periodic_flush(self): while True: time.sleep(self.FLUSH_INTERVAL) with self._lock: if self._buffer: self._flush_unlocked() def _flush_unlocked(self): if not self._buffer: return batch = self._buffer[:] self._buffer.clear() # 失败时把批次重新放回队列，下一次 flush 再试 try: resp = requests.post(self.endpoint + "/collect", json=batch, timeout=2) if resp.status_code != 200: self._buffer[:0] = batch except Exception: self._buffer[:0] = batch

这段代码有两个细节值得说。第一个是清空缓冲区后再发送，而不是发送成功后再清空，这样可以避免在网络阻塞时锁占用时间太长。第二个是失败回放：把整个批次放回队首，保证不丢消息，代价是极端情况下可能造成部分重复，由下游去重兜底。生产环境里我建议再加一个最大重试次数，超过后把数据落盘，防止内存被阻塞的网络请求耗光。

3.2 消费服务：Redis Stream 的读取与幂等写入

消费端是 rea 里逻辑最重的一环。我的实现是启动多个消费者线程，各自通过 XREADGROUP 读取 Redis Stream 中的新消息。关键代码如下：

import redis STREAM_KEY = "rea:event-stream" GROUP_NAME = "rea-consumers" def create_group(r): try: r.xgroup_create(STREAM_KEY, GROUP_NAME, id="0", mkstream=True) except redis.exceptions.ResponseError as e: # Group already exists 时可忽略 if "BUSYGROUP" not in str(e): raise def consume(r, consumer_name, handle_batch): while True: raw = r.xreadgroup( GROUP_NAME, consumer_name, {STREAM_KEY: ">"}, count=100, block=1000 ) if not raw: continue messages = raw[0][1] events = [parse(msg) for msg in messages] # 批量处理成功后，调用 xack 确认消费 handle_batch(events) r.xack(STREAM_KEY, GROUP_NAME, *[m[0] for m in messages])

这里最重要的编程纪律是：必须等 handle_batch 完全成功后再 XACK。如果先确认再处理，一旦进程崩溃，这批消息就永久丢失了；如果处理成功却没确认，消息会在下次读取时被重新投递，但相比丢失，重复是更容易接受的错误。重复问题交给下游去重，这就是典型的取舍权衡。

3.3 落库优化：批量 INSERT 与分区表设计

消费服务将清洗后的事件写入存储层时，我采用了批量 INSERT 的方式，每攒够 500 条记录或者每两秒执行一次写入。这样做比逐条插入高效得多，也减少了对数据库连接的开销。批量写入还有一个附带好处：事务边界更清晰，一次写入要么全部成功，要么全部回滚，不容易出现一半数据成功一半失败的分裂情况。

建表语句按照时间分区设计，我用的分区粒度是小时：

CREATE TABLE event_detail ( event_id VARCHAR(64) NOT NULL, event_type VARCHAR(32) NOT NULL, event_time DATETIME NOT NULL, service VARCHAR(64) NOT NULL, instance VARCHAR(64) NOT NULL, dimensions JSON NOT NULL, metrics JSON NOT NULL, PRIMARY KEY (event_id, event_time) ) PARTITION BY RANGE (TO_DAYS(event_time)) ( PARTITION p20250120 VALUES LESS THAN (TO_DAYS('2025-01-21')), PARTITION p20250121 VALUES LESS THAN (TO_DAYS('2025-01-22')), PARTITION p20250122 VALUES LESS THAN (TO_DAYS('2025-01-23')) );

分区表的意义在于查询裁剪和生命周期管理。查询时带上 event_time 条件，数据库就可以直接定位到对应分区，避免全表扫描。清理过期数据时也只需要执行一次 DROP PARTITION 操作，比 DELETE 逐条删除高效好几个数量级。我在 rea 中设定的保留窗口是最近 30 天，每天凌晨自动把 30 天以前的分区全部删掉。

4. 实战复盘：rea 在测试环境暴露的三个典型问题

4.1 消费延迟从 200ms 飙到 9 秒：消费者组的空转轮询

rea 第一个版本跑通后，我做了简单的压测。开始阶段一切正常，端到端延迟稳定在 200ms 左右。但当我启动 8 个消费线程时，延迟突然飙升到 9 秒，而且 CPU 占用率也异常高。我一度怀疑是 Redis 性能问题，但看了监控指标后，Redis 的 CPU 和内存都很平稳。

排查过程花了不少时间。我先在消费服务里加了耗时日志，发现每次 XREADGROUP 返回空结果的耗时极短，单个线程的空转几乎没有成本，但 8 个线程全都在高频循环空转时，累积消耗的 CPU 是相当可观的。随后我检查了阻塞参数 block=1000，发现问题并不在单次调用，而在于循环没有退避机制：block 超时返回 None 后，代码立即发起下一次 XREADGROUP，等于每秒钟每个线程发出了 10 到 15 次空查询。

修复思路很简单：在读取到空结果时加入 sleep 退避，或者在 block 超时后增加一个轻微延时。我把 block 时间从 1000 毫秒调整为 5000 毫秒，并在空结果分支加上了 50ms 的 sleep。修正后，8 个线程的 CPU 占用率下降了约六成，延迟也回到了正常区间。这个问题的教训是：消费者组的空转轮询看似无害，但在线程数量放大后，会凭空吃掉大量 CPU，真实生产环境一定要观察无消息时的资源消耗。

4.2 消费延迟从 200ms 飙到 9 秒：消费者组的空转轮询

这里其实是另一类问题：消费分片不均。我在测试时发现，8 个消费者线程中，有一个线程的待处理列表里始终挂着大量消息，其他线程的待处理列表却几乎是空的。现象是整体延迟不高，但偶发出现单条消息延迟数秒的情况。

我分析了 Redis Stream 消费者组的分配机制。它默认按照"按需拉取"的方式分发消息，哪个消费者先来读取就给哪个消费者分配消息，并没有做任何均衡策略。测试环境里线程启动顺序固定，导致先启动的线程总是抢到绝大部分消息。这个问题在做水平扩容时尤其隐蔽：新增一个消费者实例后，它并不会自动分担存量消息，只会从后续新增消息开始接收。

处理方案分两步。短期的做法是让消费实例定期通过 pending 列表清理自己的积压；长期的思路是尽量保持消费者实例数量稳定，避免频繁增减。在 rea 这个体量下，我更倾向于在接受轻微不均的同时，用延迟告警兜底，而不是投入精力去做精确均衡。

4.3 存储水位告急：保留策略与分区清理的连带影响

上线一段时间后，我收到磁盘告警。查了数据库后发现，event_detail 表占用了远超预期的空间。明明设计了 30 天的保留窗口，为什么空间还是涨这么快？我登录数据库查看分区情况，发现清理任务根本没有执行。

问题出在清理任务的调度上。我的实现是每个小时跑一次循环检测，理论上应该在跨天时触发 DROP PARTITION，但脚本里有一段判断分区边界的逻辑写错了：我用当前时间减去 30 天得到边界日期，再把边界日期和分区的 VALUES LESS THAN 做比较，但分区边界是每天零点，而边界日期是带时分秒的时间戳，类型不一致导致比较永远失败，于是所有旧分区都留在了表里。

修复方式不是简单改一行代码，而是彻底梳理了分区管理的自动化脚本。我把分区维护做成一个独立模块，每天凌晨检查一次需要创建哪些新分区、删除哪些旧分区，并加上试运行模式：先输出即将执行的 DDL 语句，确认无误后再真正执行。存储空间的问题解决后，我又把表空间的每日增长量加进了监控面板，防止这类问题再次悄悄发生。

另外，因为 event_detail 表使用了 JSON 字段，我在清理数据时也特别注意了索引的维护频率。大批量 DROP PARTITION 之后，索引碎片率可能会有一定上升，生产环境建议在清理任务执行后顺手做一次索引重建，避免查询性能因为碎片累积而逐渐退化。

5. 经验沉淀：rea 离可用还差什么、后续如何演进

5.1 对实时链路监控的重新认识

rea 让我重新理解了监控在实时链路中的位置。批处理时代出问题，最坏情况是报表晚出几个小时，大家顶多发个公告；实时链路一旦出问题，数据就直接呈现在业务面前，任何一点延迟、缺失或重复都会被放大。所以光把管道搭起来远远不够，必须给管道自己装上"仪表盘"。

我在 rea 中加了三类核心指标。第一类是接入层的上报量，单位时间内收到了多少条 HTTP 请求；第二类是缓冲层的积压量，通过 XLEN 查看 Redis Stream 的长度，数值持续增长就说明消费能力跟不上；第三类是消费端的处理耗时和失败率。这三类指标分别对应管道的入口、通道和出口，任何一个环节出问题都能第一时间看到大致方位。

在告警规则上，我也吸取了前面踩坑的教训。延迟告警的阈值不能设得太死，否则会频繁误报；但也不能太宽，否则失去了告警的意义。我的做法是设置两个等级的阈值：超过 3 秒触发警告，超过 10 秒触发严重告警，并配合持续时长的判断避免单次抖动惊动值班人员。

5.2 基于 rea 可以继续走的两个演进方向

如果你看完这篇文章也想搭一套类似管道，我建议下一个阶段考虑两个方向。

方向一是增加实时聚合能力。目前 rea 的消费端是清洗后就落库，查询时再去数据库做聚合。数据量继续增长后，查询响应会变慢，因为聚合计算发生在每次查询请求中。更好的做法是在消费过程中直接维护分钟级的聚合结果，每收到一条事件就更新对应分钟桶的计数，查询时直接读聚合结果而不是扫描明细表。这意味着要引入窗口状态管理，但提升是明显的。

方向二是将明细数据导出到离线数仓做深度加工。实时管道解决了"现在发生了什么"的问题，离线数仓更适合回答"长期趋势如何"。两者互补，rea 在中间扮演的是"新鲜数据搬运工"的角色。导出时要注意数据格式的统一，最好用列式存储格式，这样数仓侧的查询性能会更好。

5.3 写实时管道前，先把对账逻辑想清楚

最后分享一个我在 rea 开发中后期才彻底想明白的细节：对账。批处理时代，每天跑完批后可以用总行数校对两边数据是否一致；实时管道是连续流动的数据，根本没有一个明确的"全部跑完"的时刻，对账就变得困难。

我的做法是单独建立一个汇总任务，每十分钟统计一次 Redis Stream 中的总消息数和数据库中的落库行数，对比差异。正常情况应该完全相等，一旦出现差异，就说明中间链路丢了数据或产生了重复。配合 event_id 去重，我很快就能定位到具体是接收集中的丢失还是消费确认逻辑的问题。对账逻辑本身很简单，但它的价值是在数据异常还没被业务感知之前就发出报警。

做实时管道的这段经历让我最大的收获是：真正困难的地方并不在于某个具体组件怎么用，而在于如何把全链路的可靠性、可观测性、排查效率这些"软性要求"落实到代码里。rea 现在还很轻，但它的骨架和处理问题的思路，我觉得是值得复制到很多场景里的。