如果只给你三个字母“REA”你会想到什么在我这里它是 Real-time Event Analytics 的缩写——一套从采集、缓冲、聚合到展示的实时事件分析系统。我花了两周时间把它从零搭起来中间推翻了两次设计最终跑在了生产环境上。这篇文章是完整的复盘适合那些想自己动手做日志分析、用户行为统计或监控告警系统又不想一上来就搬整套大数据全家桶的开发者。先交代背景某团队的核心业务需要一个“分钟级”的行为洞察看板要看到实时在线人数、核心按钮点击量、页面访问趋势甚至能按渠道拆解用户在哪个环节流失。传统方案是每天晚上离线抽数第二天早上开会看报表但线上活动一开这一套就完全跟不上。大家要的是“活动上线后喝口水的功夫就能看到转化曲线”。于是REA 这个内部项目就在这种需求下诞生了。1. 先想清楚 REA 到底要解决什么问题1.1 从业务痛点倒推技术指标代码不是凭空写的指标是从业务场景里长出来的。我当时拉上业务方聊了四个关键问题数据量能到多少延迟容忍度是多少哪些指标必须实时哪些允许事后修正。这个动作非常重要因为它直接决定了后面所有技术选型的边界。最后得到的核心需求是日均事件量在 3000 万到 5000 万之间高峰在晚上八点到十点大概是日常的 5 倍。延迟目标事件发生后最多 2 分钟出现在看板上核心指标如在线人数要求 30 秒内刷新。必须支持多维度拆解事件类型、页面路径、用户渠道、设备类型。允许少量误差但不能超过 2%且关键链路如支付成功事件必须保证不丢不重。这个量级其实很尴尬。用离线数仓太重用完整的大数据实时计算框架又杀鸡用牛刀。算一笔账如果直接用 Flink Kafka HBase光运维成本就够一个小团队喝一壶。而当时团队里没人专门做大数据运维主力是写后端服务的更熟悉消息队列和关系型数据库。所以 REA 项目的设计原则很快就定了用最少的组件做一件明确的事所有组件都尽量选团队已有的技术栈。1.2 为什么选择自研而不是直接套用现有方案市面上不是没有现成的实时分析产品。我也调研过几款开源工具比如只需要部署一个 agent 就能接日志的轻量级方案还有自带 UI 的监控平台。但最后都没用原因很实际第一商业方案或重型开源方案把逻辑封得太死业务自定义维度没法做。我们要的是“用户点了哪个按钮”、“从哪个页面跳转到哪个页面”这种偏产品视角的行为分析不太适合纯基础设施类的监控工具。第二数据安全要求。业务事件日志里面有用户 ID、设备指纹这些敏感信息不能随意传到外部服务去解析。自建管道所有数据都留在自己的内网合规上可控。第三团队需要在实战中积累处理流式数据的经验。如果直接套一个黑盒后面业务发展再复杂一点可能就要请外部顾问来救火。与其这样不如在量级可控的情况下自己造一轮轮子把该踩的坑踩一遍。于是 REA 定义为一个轻量级的实时事件聚合服务从 Kafka 读事件流在内存里做窗口聚合定期写到 ClickHouse再通过查询层把结果暴露给前端看板。简单说就是数据不进重型计算框架而是用“消费者 状态 定时输出”这种最朴素的方式实现。2. 技术选型每一层都是权衡出来的2.1 事件采集为什么用日志文件而非直接 HTTP 上报采集是整个链路最容易忽略但坑最多的地方。一开始我觉得最直接的方式就是业务服务直接调 REA 的 HTTP 接口把事件发过来这样简单、实时性也好。但很快被否了原因是耦合太严重。如果每个接入的业务方都直接调用 REA 接口那么 REA 的可用性就直接影响业务可用性。而且业务服务会有各种各样的异常比如网络超时、重试、连接池耗尽这些都会把任务积压到业务线程里。更麻烦的是如果 REA 升级重启业务进程里等待响应的请求会堆积搞不好把正常的业务接口也拖垮。所以后来改成“日志文件”作为事件的中转。具体做法是业务服务在关键执行点打印结构化 JSON 日志文件由 Filebeat 监听Filebeat 把日志打到 Kafka。REA 服务只从 Kafka 消费数据跟业务进程完全解耦。这个模式的好处是显而易见的业务方不需要引入任何 SDK只需要约定好日志格式REA 的故障不会反噬业务Filebeat 本身足够稳定也能保证日志不会轻易丢失。而且日志文件天然自带缓冲业务高峰时即使 Kafka 短暂不可用日志也不会丢Filebeat 恢复后会把积压的日志补齐。事件格式我们统一成了 JSON长这样{ event: click, uid: u123456, page: /activity/2024/summer, channel: wechat, device: android, ts: 1712312345678, meta: { button: submit, product_id: sku_998 } }有一点值得注意ts字段是业务侧生成的事件发生时间而不是 Filebeat 采集时间。这个字段在做时间窗口聚合时极其关键。如果忽略它直接拿 Kafka 到消费者里的系统时间做聚合那么所有网络延迟、进程阻塞都会变成误差导致跨整点的求和乱套。2.2 缓冲层Kafka 还是 Redis Stream事件进来了先落到缓冲区。我调研了两种选择Kafka 和 Redis Stream。Redis Stream 的优势是部署简单团队已经很熟 Redis不需要额外维护 Kafka。但劣势也明显数据量一旦上来内存占用会很难看写入性能虽然高但磁盘持久化能力不如 Kafka而且 Redis Stream 的消息堆积运维起来比较冷门出了问题社区方案少。Kafka 看起来重但它解决的是“消息能不能留得住、能不能回溯”的问题。日志文件本身就是持续生产的Kafka 可以长时间保存 7 天的数据哪怕 REA 聚合程序跑挂了重启后可以从旧 offset 重新消费完全不用怕丢事件。这是 Redis Stream 不太敢保证的。所以我最后选了 Kafka。topic 命名很简单就叫user-events分区数设成 6。为什么是 6这是根据目标吞吐量和消费并行度算的。单分区顺序写入的吞吐通常能到每秒几十 MB但消费端并行度取决于分区数。我们目标事件量平均在每秒 5000 条左右峰值可能到 25000 条单条 JSON 大约 400 字节。也就是说峰值数据量约为 10 MB/s。6 个分区每个分区每秒大概消费 1.7 MB完全在安全范围内。同时6 个分区意味着消费者组里最多有 6 个 worker 并行消费这个扩容空间刚好够用。2.3 聚合存储事件明细和聚合结果分开存放很多人第一反应是“分析展示直接从 Kafka 里算不就行了”。但事实是看板需要反复查询而 Kafka 不适合直接做即席查询。所以必须有存储层。一开始我犹豫过 Elasticsearch因为日志分析大家都喜欢用它。后来在测试环境里模拟了聚合查询场景发现频率极低。一是 ES 的写入有近实时延迟二是聚合查询需要扫描大量数据性能并不理想。对比之后我选了 ClickHouse。ClickHouse 是列式存储对这一类“按维度分组计数、按时间范围求和”的分析查询来说效率远高于行式存储。而且它部署也简单单机就能跑不需要像 ES 那样搞一个集群协调。数据量在千万级时单机 ClickHouse 的查询延迟能稳在百毫秒以内。存储分为两张表第一张是明细表event_log保存原始事件用于排查问题和数据回溯。但因为明细数据量会不断膨胀所以只保留 3 天。第二张是聚合结果表event_agg按分钟粒度存聚合计数用于看板展示保留 30 天。为什么不直接用一个表因为明细和聚合的查询特点不一样。明细表要随机查单个事件聚合表则要高频按时间分组求和。如果混在一起底层 row 的数量会非常大查询性能和运维复杂度都会受影响。拆成两张表后看板只查聚合表很快分析人员要 deep dive 时再查明细表两边互不干扰。ClickHouse 建表语句大致如下CREATE TABLE event_agg ( window_time DateTime, event_name String, page String, channel String, device String, event_count UInt64, unique_visitors AggregateFunction(uniq, String) ) ENGINE AggregatingMergeTree() PARTITION BY toYYYYMMDD(window_time) ORDER BY (window_time, event_name, page, channel, device)这里用AggregatingMergeTree是为了让多次写入的相同维度数据在后台自动合并。REA 程序也会在内存里按分钟聚合后再写入但仍然存在宕机导致部分重复的可能所以表引擎选择“可合并聚合”而不是纯粹的SummingMergeTree能尽量兜底。2.4 可视化Grafana 配置 / 自建 Web可观看板这块我用的是 Grafana。主要是因为它可以直接连接 ClickHouse 数据源配置面板做起来最快。Grafana 支持 SQL 查询变量我可以直接用 SQL 查询结果来驱动下拉框比如“页面列表”直接SELECT DISTINCT page FROM event_agg不用写一行后端代码。这样上线第一版看板只用了半天时间。但也有一个限制业务方希望看到一些自定义的漏斗路径比如从首页到详情页再到支付页的转化率。Grafana 做这种需要多步串联的图表很不顺手。所以我的方案是常规指标看板用 Grafana复杂的漏斗分析单独写一个轻量级 Web 页面后端只提供一个/api/funnel接口查 ClickHouse前端用简单的图表库画漏斗图。3. 核心实现流式聚合的三个关键细节3.1 时间窗口与事件时间REA 的核心逻辑是一个 Go 写的消费程序从 Kafka 拉取事件做窗口聚合再批量写入 ClickHouse。时间窗口我选了“分钟级滚动窗口”就是每分钟切换一个新桶。每一分钟内来多少事件就累加到一个计数器里。这么做会让逻辑非常简单内存里维护一个ExpiredMap[windowStartTime] map[dimension]count每秒收到事件后查一下它属于哪个分钟窗口把对应维度的计数 1。到窗口结束时把所有值取出来写入 ClickHouse然后删除该窗口的内存数据。窗口划分的关键是“事件时间”不是“处理时间”。我举一个具体的例子某用户在网络异常时点击了一个按钮事件在浏览器端生成了但网络恢复已经是 20 秒后服务器才收到。如果按处理时间聚合这个事件就会被统计到收到时刻所在的那一分钟但业务上它应该属于点击发生的那个分钟。这个偏差在高峰期会导致整点时刻的曲线出现“延迟鼓包”——看起来流量在某一分钟暴涨其实是前几分钟的迟到事件堆过来了。解决办法是在聚合前先读事件里的ts字段按事件所属的分钟取窗口粒度。REA 程序里做窗口对齐逻辑是这样的func windowStart(ts int64) int64 { return ts - ts%60 }简单得很。但实际还要处理一个问题迟到太久的事件怎么办。我们这里定的策略是拒收超过 5 分钟的事件丢弃并告警。这样避免恶意乱序数据或低频重试事件把窗口积压得越来越多。同时后台单独统计被丢弃的最近 5 分钟事件数用来监控业务侧是否存在严重延迟。3.2 聚合服务的状态管理既然用了内存聚合就得考虑状态会不会越来越大。假设有 100 万个用户、 10 个事件名、 20 个页面、 3 个渠道每个组合都占一小块内存量级在百万到千万之间。Go 这门语言在内存管理上还算高效但如果不加限制每分钟的桶会积压到上一分钟的桶还没清完下一分钟又来了。我的处理方式是两段式清理第一段程序内维护两个当前活跃窗口的计数器映射。第二个窗口还开着的时候第一个窗口的数据已经写完了就立刻从 map 里删除。第二段每次向 ClickHouse 写入后额外跑一次runtime.GC()强制回收内存因为 Go 的内存释放是懒式的不手动触发的话高并发下内存占用会暂时虚高。实际运行后单个 worker 的内存保持在 200 MB 以内很稳定。还有一点是我并不把每个事件都存内存只存聚合后的计数。这也就是为什么叫“聚合服务”而不是“解析服务”。像用户 ID 这种需要精确去重的字段我单独用unique_visitors AggregateFunction存哈希集合而不是把所有 UID 都放内存。ClickHouse 的uniq函数会自动用 HyperLogLog 做近似去重把每个维度的用户基数内存控制在几 KB 以内。由于业务允许 2% 以内的误差这个方案完全够用。不过这里要特别提醒一个我踩过的坑Go 的 map 并发读写不是线程安全的。如果把消费线程和定时输出线程共用同一个 map不加锁运行几分钟后就会出现死锁或者数据错误。我的解决办法很简单用sync.Map或者按分区把状态拆开每个分区一个状态对象不同分区之间的状态天然隔离。后者更简单粗暴也能减少锁竞争。最终我采用了后者因为 Kafka 消费者本身就是按分区分配的一个 goroutine 只处理一个分区状态不需要跨 goroutine 共享。3.3 幂等写入与数据校正实时管道最怕的就是消息重复。Kafka 消费者可能在处理一批数据、写入 ClickHouse 之后还没提交 offset 就宕机了。重启之后它会重新消费这批数据。如果我的聚合逻辑是“每收到一条事件就 1”那这批数据就会被多算一次。解决这个问题有两个层次。第一个层次是用 Kafka 的幂等生产者和消费者配置。生产者侧设enable.idempotencetrue消费者侧设enable.auto.commitfalse手动提交 offset。这样能把“不丢”变成“至少一次”暂时还做不到“精确一次”。第二个层次是让最终写入具备可重复性。如果系统逻辑已经变成“至少一次”那就要在存储层合并重复计数。如果是SummingMergeTree重复写入会被直接累加。所以我改用AggregatingMergeTree在写入前REA 会为每条聚合记录生成一个window_time 维度组合的唯一键ClickHouse 的AggregatingMergeTree在后台会按这个键合并状态。举个例子来说明第一分钟worker A 写了window202501011200, pagehome的计数为 100。worker A 因为宕机重启重新消费第一分钟的数据又写入了一条计数为 100 的合并状态。AggregatingMergeTree会在后台把这两个状态合并最终总计数还是 100而不是 200。这背后的原理是“可合并的状态函数”它只合并计数、合并基数估算值而不是粗暴求和。所以即使重复写入最终结果也只会落到一个状态里不会重复累加。这是这个项目里比较关键的一个设计。4. 部署与调优从单机到三节点的过程4.1 资源规划与容器编排REA 的整体架构是“三台机器 一个容器编排”。三台机器分别是 Kafka 节点、ClickHouse 节点、聚合服务和前端应用的部署节点。严格来说规模很小根本不需要 Kubernetes直接用 docker-compose 就能解。之前说过团队之前没有专门的大数据运维。引入 Kafka 之后为了避免环境差异我直接把 Kafka 容器化的方式跑起来。但这里有一个教训Kafka 容器化的坑不少尤其是 broker 的advertised.listeners配置。如果配置不正确Kafka 在容器外连接时会一直报connection refused。原因很简单——容器内网 IP 和宿主机 IP 不一样Kafka 对外广播的地址需要显式配置成宿主机可访问的地址。我的 docker-compose 配置片段大致长这样kafka: image: bitnami/kafka:3.6 ports: - 9092:9092 environment: - KAFKA_CFG_NODE_ID0 - KAFKA_CFG_PROCESS_ROLEScontroller,broker - KAFKA_CFG_LISTENERSPLAINTEXT://:9092 - KAFKA_CFG_ADVERTISED_LISTENERSPLAINTEXT://your-server-ip:9092 - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS0your-server-ip:9093这段配置里的ADVERTISED_LISTENERS一定要换成宿主机实际可用的 IP 或域名。如果你和我一样是在内网环境测试直接写内网 IP 就行。这个问题是新手上手时最容易撞墙的点。ClickHouse 的部署就简单了直接跑官方容器挂个数据卷然后开放 8123 端口用于 HTTP 查询。生产上我建议把 ClickHouse 的async_insert打开因为 REA 是批量写入每一次批量提交通常包含几十到几百行的聚合结果。开启async_insert后ClickHouse 会在内存缓冲一段时间再落地写入吞吐能提升不少。聚合服务本身是一个 Go 编译出来的二进制我没有让它容器化而是直接模拟成 systemd 服务跑在宿主机上。原因很简单没有多副本需求而且二进制在宿主机的日志排查起来更直接。如果有两个 worker我就跑两个独立的进程每个进程通过命令行参数指定消费分区范围比如rea-worker --partitions 0,1,2 --group rea-group rea-worker --partitions 3,4,5 --group rea-group这种模式非常直观扇出排查时也容易定位。4.2 消费组 lag 监控与扩容实时系统上线后第一个要盯的指标不是 CPU 使用率而是 Kafka 消费组的 lag——就是“消费者已经落后了生产者多少条消息”。Kafka 的 lag 有一个标准的查法kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group rea-group --describe输出里面会显示每一分区的consumer-id和LAG。如果某个分区 LAG 持续增长说明该分区对应的 worker 处理速度赶不上生产速度。我第一次压测就碰到了这种情况。当时 6 个分区都有 lag而且越积越多。看了下日志发现 ClickHouse 在写入时遇到了慢查询单次批量写入的耗时超过了 10 秒。原因是默认情况下 ClickHouse 的 merge 任务和写入抢 CPU导致写入高峰时经常出现锁等待。排查之后做了两个操作一个是把 ClickHouse 的merge线程数调低到 2另一个是在 REA 程序里把批量提交的条数拉大到 2000 条延长批量等待时间到 5 秒。这一改ClickHouse 的写入压力大幅下降分区的 LAG 很快就归零了。处理 lag 的逻辑其实就是一个漏斗模型如果后端写入慢前端再快也没用。所以扩容最有效的动作不是加 worker 数量而是提高单批写入效率。4.3 查询性能优化物化视图和索引看板跑起来之后业务方反馈图表加载慢尤其是一些聚合维度比较多的页面加载要好几秒。查了之后发现问题出在每次看板查询都走event_agg全表扫描虽然量级不大但按天分区后仍然要扫大量行。于是我在 ClickHouse 里建了物化视图把最常用到的维度组合进一步预聚合。比如业务问得最多的是“每天每个渠道的 PV 和 UV”我就在event_agg之上再创建一个日维度的二阶段聚合表CREATE MATERIALIZED VIEW event_agg_daily ENGINE AggregatingMergeTree() PARTITION BY toYYYYMM(window_time) ORDER BY (window_time, event_name, channel) AS SELECT toStartOfDay(window_time) AS window_time, event_name, channel, sumState(event_count) AS total_count, uniqState(unique_visitors) AS unique_visitors FROM event_agg GROUP BY window_time, event_name, channel这个物化视图会在数据写入event_agg时自动更新不需要 REA 额外调用。查询日维度数据时直接查event_agg_daily而不是event_agg行数一下少了几个数量级延迟稳在 200 ms 以内。如果还想再快可以在普通字段上建minmax索引比如page字段。不过实测下来对于这个量级意义不大所以我没有额外加。5. 常见问题与排查技巧实录5.1 Kafka 消费者连续重启项目上线第一周我的一个 worker 出现周期性的重启。最明显的现象是日志打印了REBALANCE相关消息但没有任何异常堆栈。排查了一下发现是消费组里的 worker 数量大于分区数导致的。如果你把 6 个分区配置给 6 个 worker那没问题。但如果你图省事把 6 个 worker 全挂到同一个组又让它们监听相同的主题那么 Kafka 会频繁地进行 rebalance。每次 rebalance 期间消费者会暂停消费这会引入额外的延迟和重复消费风险。解决方案很简单worker 数不要超过分区数并且每个 worker 只处理它被指定的分区。这种“静态分配”模式能最大程度减少 rebalance 带来的抖动。5.2 聚合窗口边界出现“双写”我在监控时发现跨分钟整点时某些维度的计数值会略有虚高。排查了很久终于定位到问题出在windowStart计算时不严谨。原本这样写func windowStart(ts int64) int64 { return ts / 60 * 60 }在大多数时间没问题但当ts是负数或者 0 时这种除法会得到错误结果。更诡异的是如果ts是毫秒级时间戳你还得先把毫秒转成秒。我一开始没转直接把毫秒除 60导致每分钟窗口的起点都在半分钟附近的偏移量上跨整点时两个窗口都收了一条事件。后来统一改成func windowStart(tsMs int64) int64 { sec : tsMs / 1000 return sec - sec%60 }并且在整个项目里明确所有内部时间统一用毫秒时间戳直到写入 ClickHouse 时才转成DateTime。这样改完边界虚高的问题就消失了。这里想提醒大家的是时间窗口的代码看起来简单但恰恰是这一类“算术边界”最容易埋雷。建议写完之后专门写一个单测覆盖这些奇怪的边界值整点前 1 毫秒、整点后 1 毫秒、跨日、跨月。5.3 热门 key 导致数据处理不均业务上线了一个爆款活动瞬间流量冲到正常值的 8 倍。我发现 ClickHouse 写入速度没崩但是 Kafka 某个分区的 LAG 突然飙升。看了曲线其他分区 lag 几乎为零只有那一个分区积压严重。原因很简单活动页面只有一个所有点击事件都带同一个pageactivity_xxx而我按page字段做分区分配导致某个 worker 承接了绝大多数数据。这个问题的本质是“数据倾斜”。解决思路有两种一是在聚合服务内部把热点维度的计算拆成子键比如把同一个page随机打散成 16 个子键然后二次聚合。二是调整 Kafka 分区的 key 分配策略让事件按uid而不是按page分布。我一开始选择了更省事的第二种方案因为业务侧日志里本来就带uid按uid哈希到不同分区后天然能摊平热点页面的压力。但后来发现按uid分配后某些维度比如page的聚合需要在消费端再做一次“二次聚合”因为同一page被分散到了多个分区。于是我在聚合服务里增加了一个中间层每个 worker 把聚合结果写到 Kafka 的一个临时 topic然后由一层新的 worker 做汇总。这个方案虽然增加了一步但从结果看热点 key 的处理能力提升了接近 10 倍。5.4 坐在电脑前排查“数据凭空消失”的烦恼一次联调中我看板上的数据突然少了某个渠道的统计。查了很久发现是事件日志里的channel字段偶尔会出现空值。因为日志是业务方打的有些老接口并没有传渠道字段。我的心一横就想在聚合层把空值统一成unknown。这种“脏数据兜底”应该在入参处做。我的做法是在 REA 程序里对所有维度字段做非空校验空值补unknown同时把这个纠正动作记到专属日志里方便后期反查。这一改之后看板就不再有“凭空消失”的数据了。6. 一点个人体会REA 这个项目谈不上多高大上它没有用 Flink没有用复杂的数仓分层也没有上 K8s 做完整微服务治理。但它非常贴合当时的业务规模和团队能力。很多开发者一谈到实时分析就想到必须上全套流式计算框架其实完全没必要。先想清楚数据量、延迟、可靠性这三个核心指标再倒推技术选型你会发现很多“轻量级”方案反而更长久。其中最让我受益的一个习惯是每做一个技术决策都写下放弃的备选方案。比如当时我为什么没有用 Redis Stream为什么没有直接用 ES 看图为什么不去改业务代码而改用日志采集。这些决策理由写下来后后续团队接手 REA 时不用重新踩一遍讨论过程直接看文档就能理解系统边界。最后分享一个维稳的小技巧把所有跟时间有关的字段全部打印成毫秒时间戳并且在日志里附上 Kafka 分区和 offset。这样一旦出现数据重复或者丢失你能用最直接的线索回溯到“是哪一批数据、哪个分区、哪个 offset 出了问题”。REA 跑了大半年这个习惯帮我省下了无数次熬夜排查的成本。如果这个项目后续有继续扩展的想法我可能会把聚合服务抽出来做成独立的数据处理插件让更多业务方按自己的维度接入把 REA 的“事件分析”能力沉淀成团队内部的基础平台。但现在这套轻量级方案已经稳稳扛住了业务高峰这就够了。