轻量级事件分析服务rea：基于FastAPI与SQLite的工程实践

项目代号就叫rea三个字母别想太多。第一次把这个名字发给朋友看对方第一反应是“React 的亲戚”——说实话它不是前端框架也不是某个开源库的缩写变体。我用这个名字做了一个轻量级的事件分析服务全称取Rapid Event Analytics的缩写三个字母好记也刚好和“Real”的前缀擦边算是给自己一点心理暗示这东西要足够真实、足够落地。做这个项目的起因很朴素当时被一个小团队拉去搞内部业务埋点需要把散落在几个服务里的业务事件统一收上来既要做实时统计也要支持事后按条件回查。评估了一圈现有方案之后我不想为一个小需求引入一整套分布式组件决定自己写一个够用且不折腾的方案。rea 能做什么一句话说清楚接收 HTTP 事件上报写入本地存储提供按时间窗口的聚合查询和明细检索。它适合小团队、个人开发者、边缘设备或者任何“数据量没那么大但就是想要一个顺手工具”的场景。这篇文章我会把 rea 从需求分析、架构设计到核心代码、部署压测、踩坑记录完整过一遍所有代码都是可以直接抄走的水平。老读者都知道我不太爱讲虚的能写的细节全部拉满希望你读完能少走几段弯路。1. 项目定位与设计取舍1.1 需求起点被“重组件”逼出来的自研当时那个团队的情况是这样的几个内部服务各自打印日志有 access log也有业务埋点事件比如用户点击、订单状态流转、定时任务执行结果。这些日志分散在不同机器上出了问题只能登录服务器 grep想统计某类事件的 PV 曲线更是无从下手。他们的诉求很明确每天大概几万到几十万条事件不需要精确到毫秒级的实时性但希望延迟控制在秒级以内要能按事件类型和来源快速过滤要能偶尔翻一翻历史明细。最直接的路子是把 Kafka 拉起来配合流处理引擎做实时聚合再加一个时序库存历史数据。但这对运维能力是有要求的Kafka 集群要人看着流处理任务要有状态管理时序库又占内存。一个十万级日活的产品为这点数据量引入这么重的技术栈后续每一次版本升级都是折磨。更重要的是他们的开发环境和生产环境都很简陋连容器平台都是最近才开始试水。所以我的判断很明确与其去踩大组件运维的坑不如写一个单机部署、自动落盘、能跑几年都不用管的小服务。1.2 为什么是“rea”而不是现成开源件网上能搜到不少事件收集和聚合的开源项目但真正能满足“轻量、简单、无外部依赖”这几个条件的并不多。我拉过一个对比表直接把需求摆上去看方案优点主要问题适合场景Kafka 流处理引擎吞吐高、生态全部署和运维成本高最少三节点才踏实大规模、跨团队的日志/事件中枢ClickHouse 直写查询性能强、列式压缩需要独立服务表结构和分区要设计偏分析型、维度较多的数据InfluxDB / 时序库自带聚合函数、看板配套新增一个技术组件内存占用偏高传感器、监控指标类数据直接写 ELK搜索能力强、界面友好组件多、资源吃紧小项目有点奢侈全文本日志检索rea自研单文件存储、HTTP 接口、代码可控功能边界有限不支持水平扩展中小规模业务事件、边缘端采集表格一拉出来答案其实已经很明显。项目要的不是“业界标准架构”而是“刚好够用、坏了能自己修”的东西。rea 的定位就是填补这个空档没有外部中间件没有独立的客户端 SDK只要会发 POST 请求就可以接入事件上报。后续哪怕真到量级兜不住的时候把接收层换成消息队列也不算伤筋动骨。1.3 边界与取舍什么该做什么不该做好项目都有一个共同点知道自己不做什么。rea 在一开始就划了几条线。第一不做分布式。数据文件存在单机上服务挂了重启就是数据不丢是底线。跨多机聚合这种需求等真的出现再说。第二不保证全局有序。事件时间以业务方上报的 event_time 为准不上来自己打时间戳。这样即使网络有抖动后面做时间窗口聚合也不会被入库时间干扰。第三不存无限大的 payload。业务方传的附加信息 JSON默认限制在 8KB 以内。超过直接截断避免单条事件把存储打爆。实际跑下来绝大多数业务事件连 1KB 都用不到。这个边界感挺重要的。很多自研项目死于过度设计因为一开始就想着以后什么都能干结果代码越写越复杂最后连启动都费劲。我把 rea 定位成一个带存储的计算器不是全功能的 ERP反而让它跑得很稳。2. 整体架构与数据流设计2.1 数据链路接收 → 队列 → 批量落盘 → 查询rea 的数据流是一条很直的管道一共四段客户端 POST 事件 → FastAPI 接口校验 → 内存队列 → 后台线程批量写入 SQLite → 查询接口按需聚合。这里最值得解释的是为什么要在接口和数据库之间夹一个内存队列。刚开始我也想过“来一条写一条”逻辑最简单但马上意识到问题SQLite 单条写入要开启事务每条都 fsync 的话高频时性能会很难看而且大量并发写会触发锁竞争。加一层队列之后接口只需要做参数校验和入队耗时骤降后台消费者线程攒够一批或到时间阈值再统一写库事务数量少了两个数量级写锁冲突也大幅缓解。打个比方这就跟食堂打饭一样。如果每个学生都挤到窗口要求现炒整个队伍直接瘫痪现在前面有一个备餐台师傅提前把菜装盘学生来了端走就行窗口只负责快速分发。接口和存储之间的队列就是这个备餐台。2.2 存储选型SQLite 为什么够用很多人一听 SQLite 就觉得是手机或嵌入式项目才用的东西这是误读。SQLite 是单文件数据库不假但它完全具备处理中小规模服务器的能力关键是得会用。rea 选 SQLite 有几个硬理由。一是零运维没有单独进程没有端口没有用户权限体系数据就是一个.db文件备份直接拷贝文件即可恢复也是拷回去。二是事务可靠默认情况下的耐久性和并发控制足够应付业务事件场景。三是在 WAL 模式Write-Ahead Logging下读写可以并行读不会阻塞写写也不会阻塞读这正好命中“一边持续写入、一边随时查询”的需求。开启 WAL 的写法很简单库连接初始化后执行一条 SQLPRAGMA journal_modeWAL; PRAGMA busy_timeout5000; PRAGMA synchronousNORMAL;synchronousNORMAL在 WAL 模式下是推荐组合兼顾性能和数据安全。busy_timeout5000意思是如果遇上锁最多等 5 秒而不是立刻报错。这几个 PRAGMA 设置是 SQLite 能不能扛住写入压力的关键。要说局限也很明显单机写入吞吐在普通磁盘上大概能到每秒一两千条如果目标是每秒几万条事件那 SQLite 就不合适了。但 rea 的定位本来就是中小规模这个边界我认。2.3 事件模型设计rea 的事件模型是一个扁平的 JSON包含五个核心字段event_id业务方生成的唯一 ID用于幂等去重。event_time业务时间ISO 8601 格式统一用 UTC。source来源标识比如服务名或设备 ID。event_type事件类型比如order.created、task.finished。payload附加数据的 JSON 对象最多 8KB。很多人在设计事件模型时会犯一个错误把所有业务字段都拆成独立列。比如“用户 ID”一列、“商品 ID”一列、“渠道”一列。这样做确实查询方便但代价是每加一个业务字段就要改表结构事件模型就失去了扩展性。rea 的做法是把通用维度抽出来建列把差异化数据统一塞进payload查询时用 SQLite 的json_extract函数处理。这样既保留灵活性又不至于需要在 2 点紧急上线的时候改表加列。event_id这个字段常被忽略但实际价值巨大。网络请求偶尔会重试如果没有幂等键同一条事件就会写两次后续统计数字全是错的。rea 在入库时对event_id建唯一索引重复上报直接忽略从源头解决重复问题。3. 核心实现与关键代码3.1 环境准备与初始化rea 用 Python 3.10 写的依赖非常克制只有 FastAPI 和 Uvicorn数据库直接用 Python 自带的sqlite3没引入 ORM。我始终觉得这种小工具没必要上 SQLAlchemy自带库足够用还少一层依赖。mkdir rea cd rea python3 -m venv venv source venv/bin/activate pip install fastapi uvicorn初始化数据库的脚本如下注意索引的设计import sqlite3 def init_db(db_pathrea.db): conn sqlite3.connect(db_path) conn.execute(PRAGMA journal_modeWAL;) conn.execute(PRAGMA busy_timeout5000;) conn.execute( CREATE TABLE IF NOT EXISTS events ( id INTEGER PRIMARY KEY AUTOINCREMENT, event_id TEXT NOT NULL, event_time TEXT NOT NULL, source TEXT NOT NULL, event_type TEXT NOT NULL, payload TEXT NOT NULL, received_at TEXT NOT NULL DEFAULT (datetime(now)), UNIQUE(event_id) ); ) conn.execute( CREATE INDEX IF NOT EXISTS idx_time_type ON events (event_time, event_type); ) conn.execute( CREATE INDEX IF NOT EXISTS idx_source ON events (source); ) conn.commit() conn.close()idx_time_type这个复合索引是查询的关键。后面按时间范围过滤某类事件时可以走索引快速锁定数据而不是全表扫描。这个索引一定要建等数据量到几十万条的时候有没有索引的查询速度差距是几十倍的量级。3.2 事件接收接口接收接口的逻辑非常直接校验参数、写入内存队列、返回 200。在正常情况下这个接口的响应时间应该控制在 5 毫秒以内因为真正落盘的活儿已经由后台线程接管了。from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field from datetime import datetime import queue import uuid app FastAPI() event_queue queue.Queue(maxsize10000) class EventPayload(BaseModel): event_id: str Field(default_factorylambda: str(uuid.uuid4())) event_time: str | None None source: str Field(min_length1, max_length64) event_type: str Field(min_length1, max_length128) payload: dict Field(default_factorydict) app.post(/api/v1/events) async def receive_event(evt: EventPayload): if evt.event_time is None: evt.event_time datetime.utcnow().strftime(%Y-%m-%dT%H:%M:%S.%fZ) try: event_queue.put(evt.model_dump(), blockFalse) except queue.Full: raise HTTPException(status_code503, detailqueue is full, retry later) return {status: ok, event_id: evt.event_id}队列上限设置为 10000一旦积压超过这个值说明消费端处理不过来。这种情况直接返回 503 比默默吞掉数据更合适。客户端收到 503 可以做重试或降级而如果服务端假装成功数据就悄无声息丢了。使用 Pydantic 做校验省了不少事。payload必须是一个 JSON 对象如果业务方传了数组或字符串接口会自动给出格式错误提示不需要手写try-except去解析。3.3 队列消费与批量落盘后台消费者线程是 rea 的核心负责把内存队列里的数据批量写入 SQLite。import sqlite3 import time import threading BATCH_SIZE 500 FLUSH_INTERVAL_SEC 1.0 def consumer_loop(): conn sqlite3.connect(rea.db) while True: batch [] deadline time.time() FLUSH_INTERVAL_SEC while len(batch) BATCH_SIZE and time.time() deadline: try: item event_queue.get(timeout0.2) batch.append(item) except queue.Empty: if len(batch) 0: break time.sleep(0.05) if not batch: continue _write_batch(conn, batch) def _write_batch(conn, batch): try: conn.execute(BEGIN) conn.executemany( INSERT OR IGNORE INTO events (event_id, event_time, source, event_type, payload) VALUES (?, ?, ?, ?, ?) , [ ( evt[event_id], evt[event_time], evt[source], evt[event_type], json.dumps(evt[payload], ensure_asciiFalse), ) for evt in batch ], ) conn.commit() except Exception as exc: conn.rollback() # 打印日志到 stderr方便排查 print(exc, filesys.stderr) threading.Thread(targetconsumer_loop, daemonTrue).start()这段代码有两个细节值得说。第一INSERT OR IGNORE配合前面建的唯一索引起到了幂等过滤的效果。同一事件即使网络重试导致入队两次最终也只有一条落到库里。第二批量策略用的是“数量和时间的双阈值”。500 条或 1 秒哪个先到就触发写入。这样做的好处是在低峰期事件也不会在队列里等太久最迟一秒钟就会落盘而在高峰时每 500 条写一次大幅减少了事务开销。你可能注意到了write_batch里还手动包了一层BEGIN。SQLite 的 Python 驱动默认会在执行 insert 时自动开事务但手动控制事务边界更放心尤其是批量执行时一个事务完成所有写入错误回滚也干净利落。3.4 时间窗口聚合查询事件存进去就是为了查。rea 提供两类查询按时间窗口聚合和明细筛选。聚合查询接口设计成可以指定时间范围、粒度分钟、小时、天、事件类型和来源SQL 动态拼接app.get(/api/v1/events/stats) async def stats( start: str, end: str, step: str minute, source: str | None None, event_type: str | None None, ): interval_map {minute: %Y-%m-%dT%H:%M, hour: %Y-%m-%dT%H, day: %Y-%m-%d} time_fmt interval_map[step] conditions [event_time ?, event_time ?] params [start, end] if source: conditions.append(source ?) params.append(source) if event_type: conditions.append(event_type ?) params.append(event_type) sql f SELECT strftime({time_fmt}, event_time) AS bucket, event_type, COUNT(*) AS cnt FROM events WHERE { AND .join(conditions)} GROUP BY bucket, event_type ORDER BY bucket ASC with sqlite3.connect(rea.db) as conn: rows conn.execute(sql, params).fetchall() return [{bucket: r[0], event_type: r[1], count: r[2]} for r in rows]这个接口是出图表的核心依赖。前端拿到按分钟分组的数据直接就能画折线图或柱状图。strftime是 SQLite 内置的时间格式化函数用它对字符串时间做截断天然支持按分钟、小时、天做聚合不用在应用层做二次处理。明细查询也很直接按条件过滤然后分页app.get(/api/v1/events) async def list_events( source: str | None None, event_type: str | None None, limit: int 100, offset: int 0, ): conditions [] params [] if source: conditions.append(source ?) params.append(source) if event_type: conditions.append(event_type ?) params.append(event_type) where_sql fWHERE { AND .join(conditions)} if conditions else with sqlite3.connect(rea.db) as conn: rows conn.execute( fSELECT event_time, source, event_type, payload FROM events {where_sql} ORDER BY event_time DESC LIMIT ? OFFSET ?, [*params, limit, offset], ).fetchall() return [ {event_time: r[0], source: r[1], event_type: r[2], payload: r[3]} for r in rows ]明细查询可以加上start、end参数做时间范围过滤代码逻辑一样根据参数拼接条件即可。3.5 实时看板的最小实现聚合并不能直接“看”如果每次统计都要手写接口再配一个可视化工具对轻量项目来说反而麻烦。rea 内置一个最简易的 dashboard 页面用 HTML 加原生 JavaScript 实现请求刚才的 stats 接口画出最近 30 分钟内各事件类型的趋势。这个页面不引入任何前端框架就 80 行代码重点是把数据展示出来而不是造一个重前端。具体实现是每隔 15 秒请求一次接口用fetch拿 JSON动态生成折线数据绘制用 Canvas。代码我就不贴全量了核心思路就是定时轮询加简单绘图最重要的是图表能挂在:8080/dashboard这个路径上内部团队直接打开就能看到实时情况不需要再装 Grafana。这个轻量看板帮我省了不少事。以前给团队演示数据接入效果总要打开各种管理页面现在在浏览器里敲一个地址就行。如果你嫌 Canvas 麻烦也可以直接输出图表插件能识别的 JSON 格式前端任选。4. 部署、压测与问题排查4.1 部署方式Docker 一行启动项目这么轻部署也不能重。Dockerfile 总共十几行FROM python:3.10-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY app.py . EXPOSE 8080 CMD [uvicorn, app:app, --host, 0.0.0.0, --port, 8080, --workers, 1]启动命令docker build -t rea . docker run -d --name rea-server -p 8080:8080 -v $(pwd)/data:/app/data rea注意两个细节。一是必须用--workers 1。原因后面细说在 SQLite 模型下多进程会把简单的队列机制拆散起不到批量写库的作用。二是数据目录要挂载出来。很多人第一次跑 Docker 容器重启后数据全没了才发现没挂 volume这种低级错误能免则免。如果你不想用 Docker也可以直接扔到 systemd 里托管把uvicorn命令写成 service 文件。关键在于保证退出时能有机会把内存队列里的剩余事件刷完所以要在停止服务前给一个缓冲时间。4.2 简单压测rea 能扛多大流量我在一台普通配置的虚拟机上做了压测配置就是 2 核 4G 内存的云主机磁盘是普通的 SSD。压测工具用的是 Apache ab模拟持续 POST 请求ab -n 10000 -c 50 -p payload.json -T application/json http://localhost:8080/api/v1/events压出来的结果大致是每秒能稳定处理 600 到 1000 个事件请求请求响应时间 P99 在 30 毫秒左右。内存占用稳定在 150MB 以内CPU 占用在两核之间波动没出现明显飙高。这个数据是我个人环境里的实测结果不代表所有机器都一样。但即便在这个指标上打个对折每天 500 万条事件也稳稳接得住。对一个小团队的业务埋点来说这个量级完全够用。如果真到了不够用的那天优先扩展的方向不是把 SQLite 换成 MySQL而是水平拆分——按业务线拆多个 rea 实例互不相干。4.3 常见问题排查速查表跑了一段时间团队里陆续遇到一些问题。我把常见问题整理成了一张速查表遇到类似情况可以直接对号入座现象可能原因解决方式接口偶发返回 503队列积压触发上限提高maxsize或检查消费线程是否退出服务重启后事件变少队列中的数据未落盘就被杀在退出前等队列清空或用双缓冲队列写入报 “database is locked”SQLite 写锁冲突开启 WAL设置busy_timeout检查是否有长事务stats查询越来越慢缺少时间索引确保idx_time_type存在必要时按月分表统计数据偏多事件被重复上报检查event_id是否唯一是否漏加幂等逻辑Docker 重启后数据丢失数据目录没挂载 volume挂载-v到容器外部目录这张表看着简单但每一条都是实际踩过之后才总结出来的。尤其是“database is locked”这是 SQLite 项目里最常见、最容易劝退新人的报错。其实解决办法在上面已经写清楚了WAL 加 busy_timeout再配合批量写入基本不会再遇到。5. 踩坑记录与最后一点经验5.1 其实最坑的不是 SQLite是“退出时机”很多人觉得 SQLite 的写入锁是最大的坑但我在实际使用中发现真正造成数据丢失的往往是服务的退出时机关不。某次我直接停容器第二天对账发现少了几百条事件排查逻辑绕了一大圈最后定位到问题内存队列里的数据还没刷进数据库进程就被 SIGTERM 干掉了。那个时间段恰好有个促销活动事件量突然涨到平时的五六倍队列瞬间积压停服没等消费完数据全没了。解决办法是在退出前做优雅关闭。FastAPI 里可以用on_event(shutdown)回调在里面设置一个停止标志让消费线程把队列剩余数据全部刷完再退出再给一个最长等待时间兜底。Uvicorn 收到停止信号后先等这个回调执行完毕再关闭进程。加上这段逻辑后我再也没有因为重启丢过数据。5.2 time 字段统一 UTC少一半交通事故rea 的event_time从一开始就强制使用 UTC 字符串。有几个业务方接的时候嫌麻烦想用自己的本地时间。我态度很硬上报接口只认 ISO 8601 带时区偏移或纯 UTC显示层要换算本地时间是前端的事存储层必须口径一致。后来出现过一次因为某服务传了本地东八区时间而其他服务传 UTC导致统计曲线在每天 8 点附近出现诡异凹陷的“事故”。排查了半天最后就是时区混用导致的。从那以后所有接入口都在文档里加了醒目的 UTC 要求。5.3 一点心得先解决眼前问题再考虑诗和远方rea 从写第一行代码到稳定运行前后不到一周。对比同类开源方案它可能不够“专业”但它真正解决了一个具体团队的痛点而且至今没有出过严重故障这比技术栈华丽更重要。写这套系统的过程也让我重新确认了一个经验在没有明确量级预期的情况下能用单文件解决的不要上集群能用简单队列解决的不要上消息中间件。最后再分享一个小操作。如果你要用 rea 存跨月的数据建议每个月新建一张表表名带月份后缀比如events_202501这样查询时直接指定月份减少单表膨胀。这个逻辑不复杂但对远期维护帮助巨大顺手加上不吃亏。