前阵子一直在忙一个代号叫 financial-services 内部项目的重构这个项目说白了就是一套面向金融服务场景的通用数字化服务平台。不是那种给人一种高深莫测感觉的量化交易系统也不涉及任何复杂金融模型它更像是一套“运营中台”——负责把分散在不同业务方的行情数据、用户行为数据、账务流水数据统一收进来清洗、计算、存储再通过规则引擎和接口服务输出给前端展示、风险识别和业务决策使用。我之所以想把这次实践经验完整写出来是因为这个项目踩了不少坑也沉淀了一些通用方法论。不管你是做后端开发、数据工程还是刚到金融类公司做技术平台多多少少都能从中找到一些参考。尤其是一些架构选型的取舍、实时指标计算的细节以及生产环境里常见的脏数据和延迟问题这些都不是参考文档能告诉你的。如果你正准备搭一套类似的内部数据服务平台那这篇文章应该能帮你少走几条弯路。1. 项目背景与整体设计思路1.1 这个项目到底要解决什么问题先说背景。业务侧有多个渠道包括移动端、网页端、合作方接口都会产生大量的服务数据。过去这些数据各自为政业务A关心交易量业务B关心用户活跃度业务C关心异常波动每个团队各自拉数据、各自建表、各自开发报表最后口径对不上指标算出来的结果经常相差百分之二三十领导问起来谁也说不太清楚。financial-services 这个项目最初的目标非常朴素把数据统一收口建立一个稳定的指标计算平台。但做着做着发现光“统一收口”还不够因为业务方对实时性的要求越来越高。以前跑T1的离线报表还行现在业务方希望一个活动上线后能实时看到用户参与情况、交易成功率、异常波动预警。这就把项目的技术重心从“批量处理”推向了“实时/准实时计算”。所以最终这个项目的核心定位是三块数据接入标准化、指标计算统一化、预警输出自动化。听起来不复杂但真正落地时每一块都有不少细节。1.2 技术选型的思考过程选型永远不是选最火的而是选最适合团队维护的。因为初期调研的时候我也考虑过直接上完整的流式计算框架比如 Flink甚至也想过用 ClickHouse 做实时数仓。但最后综合团队情况选了一条更务实的路线Kafka 作为消息管道统一接入各类数据源一套自研的指标计算服务负责做清洗、聚合和规则判断结果写入 MySQL 和 Redis对外提供 REST API 和 WebSocket 推送。为什么没有用 Flink其实也不是不能而是学习成本和运维成本的问题。团队当时对 Java 和 Spring Boot 非常熟悉但对 Flink 的部署、调优、状态后端管理并不熟练。对于这样一个不算超大规模的指标计算场景用 Kafka 解耦自研聚合服务完全能支撑业务量而且出现问题时团队自己能快速排查。我个人的原则是在满足性能指标的前提下尽量选择团队有掌控力的技术栈不要为了简历好看去引入后续没人维护得动的东西。当然这套设计也有明显的短板。自研计算服务的水平扩展性不如 Flink 这类专业框架如果未来每秒写入量达到几十万条就不得不重构。但考虑到业务增长曲线这个风险在接下来两年内是可控的而且我也在设计上做了分片处理留了一条扩展的后路。1.3 架构的整体分层一个容易被忽略的点是架构图看着简单真正落地时要区分清楚数据分层和业务分层。数据层面我分了四层采集层、传输层、计算层、服务层。采集层由各业务线通过 SDK 或 API 上报数据传输层是 Kafka负责削峰填谷计算层是核心承载数据清洗、指标计算、规则匹配服务层对外提供查询接口和消息推送。业务层面则是围绕“指标”来组织而不是围绕“数据源”来组织。比如用户活跃度这个指标它可能来源于用户埋点、登录日志、接口调用日志等多个数据源。如果按数据源组织逻辑后期指标扩展时会非常痛苦按指标组织逻辑新增数据源只需要在接入层做一次适配后续所有指标都能自动复用。这个设计看起来是架构层面的小事但其实决定了后续所有开发效率。我还做了一个很关键的决策计算服务内部分了两条链路一条是实时链路一条是准实时链路。实时链路处理延迟敏感类的指标比如今日交易总额、当前在线用户数秒级刷新准实时链路处理需要跨时段对比的指标比如环比增长率、异常波动检测分钟级计算。这样做的好处是不会因为少数复杂指标拖慢整体实时性也让开发时可以按需取舍计算深度。2. 核心模块拆分与细节解析2.1 数据接入层管好格式才能少踩坑数据接入是整个平台的地基也是最容易出现脏数据的地方。接入层我采用了 Topic 加版本号的策略而不是简单一个 Topic 走天下。每个业务方上报数据时必须带上 schema 版本号没有带的一律进“待修复队列”通过一个管理后台让业务方自己去查看和修正。前期会有些阵痛业务方觉得怎么这么麻烦但到了后期做指标对齐时你就会发现这个约束实在救了大命。协议上当时也纠结过是用 JSON 还是 Protobuf。坦诚讲JSON 在调试时非常友好但做复杂嵌套结构的时候校验成本太高解析性能也一般。最终用了 JSON JSON Schema 校验而不是直接上 Protobuf。原因是团队当时人手不足Protobuf 的代码生成和版本管理需要额外投入而 JSON Schema 允许业务方在管理后台通过配置来调整校验规则灵活度高很多。等业务量上来之后再考虑迁移到 Protobuf也是一个可行的演进路径。接入层的另一个隐藏细节是“幂等”。Kafka 提供了至少一次投递语义这在数据上报场景下意味着重复消息几乎一定会出现。我们在接入层生成了一个针对消息体内容的 hash 值存在 Redis 里做去重过期时间设置为 24 小时。这个处理看似简单但极大地减少了后续计算层的重复计数问题尤其是做交易量这类累加型指标时没有幂等处理数据会有很多水分。2.2 指标计算层窗口、口径与聚合一锅端指标计算层是平台的大脑也是代码量最集中的地方。我在这一层实现了一套轻量级的“指标引擎”核心思路是声明式配置每个指标由三个要素定义——数据源过滤条件聚合逻辑。数据源决定去哪些 Topic 取数过滤条件决定哪些记录参与计算聚合逻辑负责怎么算。业务方新增一个指标时只要走配置中心提交一份定义计算服务会自动编译成可执行的算子链不需要重新发版。可能有人会问为什么不用现成的流计算平台或者直接用 SQL 去定义指标其实在前期调研时SQL确实是备选方案但实际业务中有很多非标准化的逻辑例如“剔除内部测试账号”“排除特定来源的重复请求”这些用SQL表达比较绕。把指标计算留在代码里用配置驱动反而最灵活。窗口计算是这个模块里最难啃的骨头。实时指标往往涉及到滑动窗口比如“过去5分钟的交易成功率”。要实现这个我用 Redis 的 ZSET 按时间戳存储每个数据点每次计算时通过 score 范围查询取出窗口内的数据再批量执行聚合。这个方案在数据量不大时性能很好看但一旦窗口内数据量突破几十万ZSET 的操作延迟就会明显上升。后来我做了采样策略对于高吞吐的指标在写入 Redis 之前做一次预聚合把单条消息聚合成秒级的计数再把秒级计数存入 ZSET。这个优化直接将窗口查询的数据量减少了两个数量级。聚合一端还需要特别注意“重算”的问题。消息延迟乱序在分布式场景下是常态比如一条 10:00:01 的数据可能 10:01:30 才到。如果你在窗口关闭的瞬间就固化结果这条迟到的数据就永久丢失了。我对每个指标配置了一个“允许延迟时间”默认 30 秒。窗口关闭后先出一版结果延迟时间到期后再做一次修订计算如果结果有变化推送一条修订消息给下游。这种“先初算、后修正”的模式在业务容忍范围内是一个极其实用的折中方案。2.3 规则引擎与预警中心让异常自动被发现业务方对平台的一个核心期望是“别让我们天天盯着报表异常了你得告诉我们”。这就需要一个简单但高效的规则引擎。我这里没有引入 Drools 这类重量级规则框架而是实现了一个基于 Groovy 脚本的轻量级规则沙箱。每条规则由两部分组成触发条件一段表达式动作推送消息、打标记、调用回调接口。规则配置支持灰度发布可以先让 10% 的流量实验观察效果没问题再全量开放。规则容易写但预警风暴难防。最常见的情况是某个指标在短时间内抖动规则频繁触发业务方收到几十条同一个原因的告警慢慢就会产生“告警疲劳”把推送直接屏蔽掉。踩过这个坑之后我特意实现了在不同规则上的灵活防护策略一是预热规则触发后要连续 N 次持续命中才真正发出预警二是冷却同一规则两次预警之间必须至少间隔 5 分钟三是聚合同一对象的多条预警合并成一条摘要摘要里附带最近一段时间的趋势数据。这样调整后的实际效果告警量减少了约 60%业务方反而更愿意处理预警信息了。规则引擎里还有一个容易被忽略的能力动态阈值。有些指标在业务高峰和低谷时的波动差异很大如果阈值写死不是误报就是漏报。我的方案是让平台自动维护每个指标的历史置信区间以过去7天同一时间段的均值加减三倍标准差作为动态基线规则配置时可以引用这个基线。比如可以设定“当指标值低于基线下限的20%时触发预警”这个能力上线之后异常识别的准确率提升明显。2.4 对外服务层既要查得到也要推得动平台对外提供的服务主要有两类查询类 API 和实时推送通道。查询类 API 相对常规基于 MySQL 和 Redis 的一二级缓存模式热点指标走 Redis历史明细走 MySQL。这里有一个经验值得分享对于指标查询接口我强烈建议不要在 API 层再引入复杂的OR映射框架直接用一个轻量的查询对象组装 SQL 即可。因为指标查询基本都是固定模式的聚合查询过度设计只会增加排查问题的难度。实时推送这一块用的是 WebSocket配合一个轻量的推送网关。网关维护着每个业务方的订阅列表当指标计算结果有更新时把变更记录推给所有订阅了该指标的服务端连接。为了不让推送消息成为瓶颈推送频率做了限流同一指标每秒最多推送一条合并后的快照数据。实际运营下来推送最考验的不是吞吐而是连接状态的维护——客户端断线重连、心跳超时这些看似简单的问题在实际联调中比写核心逻辑还磨人。查询 API 还有一个磨人的点就是“指标对齐”。同样的“今日总交易额”不同端展示不一样会让人对平台失去信任。所以我在服务层做了一个快照表每五分钟生成一次所有核心指标的全局快照所有查询默认读取快照而不是实时精确值。这样不同服务方看到的数据是一致的只是有最多五分钟的延迟。对于对实时性要求更高的看板提供实时接口但会明确标注是“实时值”避免混用。3. 实操过程从0到1的实现路径3.1 整体技术栈与项目结构整个项目后端使用 Java 17 Spring Boot 3.x消息中间件用 Kafka 3.x缓存用 Redis 6.x存储用 MySQL 8.x。布式环境那就不用想了刚开始是在单机环境开发调试之后通过 Docker-Compose 部署到三台测试服务器上。项目代码本身按 Maven 多模块组织大致结构如下financial-services ├── financial-common # 公共代码工具类、统一返回结构、异常定义 ├── financial-collect # 数据接入服务接收HTTP上报、校验、写Kafka ├── financial-compute # 指标计算服务消费Kafka、清洗聚合、写Redis/MySQL ├── financial-rule # 规则引擎服务加载规则、执行判定、触发动作 ├── financial-api # 查询与推送服务REST API WebSocket推送 └── financial-admin # 管理后台指标配置、数据修复、规则配置对于刚开始做类似平台的同学我建议即使项目很小也尽量按这种纵向切分来组织模块。因为接入、计算、规则、服务四块的部署频率完全不同混在一个单体应用里一次计算逻辑的改动可能还要连带推送服务重新发布非常容易引发线上事故。3.2 接入与校验链路的实现接入服务本质上是替业务方扛住所有“可能出问题”的地方。业务方只需要按约定把一个数据结构 POST 过来剩下的工作全部在 financial-collect 里完成。数据接入的代码核心分三步接收请求、schema 校验、投递到 Kafka。第一步比较简单直接用一个 Controller 接收 JSON 字符串不急着反序列化成强类型对象因为后续字段可能变化强类型反而限制灵活性。第二步是校验用预加载的 JSON Schema 对消息体做校验失败的进入修复队列并通知业务方。第三步是投递这里关键在于为每条消息生成业务主键式的去重ID把来源标识、业务ID、发生时间拼接后做 MD5。实现时想说一个前面提到的坑接入层在处理量突然翻倍时会遇到线程池打满导致请求超时的情况。常规思路是调大线程池但真正的问题是 Kafka 生产者发送后等待确认的时间过长。后来我在代码里把 batch.size 从默认的 16KB 调到了 64KBlinger.ms 调整到 10mssend 的吞吐量明显上涨CPU 反而下降了。这种调优不复杂但需要你在压测时关注生产者的各项指标曲线。3.3 指标计算的代码骨架计算服务的代码骨架是消费 Kafka 消息、按指标路由、执行过滤条件、执行聚合算子、写入结果。这个骨架本身不新奇但有几个细节我认为值得分享出来。第一是消息路由。我收到了业务方上报的消息并不直接按照消息类型去查表而是在消息的 header 里塞入了动态类型标记比如metric-type: user_active消费端拿到后直接按这个标记定位到预先注册的计算单元省去了解析嵌套JSON后再判断类型的损耗。第二是过滤条件的处理。过滤条件我实现成一个 Python 风格的表达式引擎引入一个轻量级表达式解析库配置人员可以在管理后台写类似channel ! internal_test and amount 0这样的表达式保存后会编译成一个可执行的对象每次消息到达时在内存中执行。这样既摆脱了硬编码又不会因为频繁拼接字符串而误伤性能。第三是聚合算子。聚合算子根据指标类型分成 Sum、Count、Avg、DistinctCount、MaxMin 五类。每一类算子都有一个对应的装饰链比如 Counter 算子会先经过去重器再到累加器再到窗口存储。这种装饰链模式的好处是新增一个指标类型时不需要改聚合核心只要在配置里指定算子链即可。下面是计算服务消费并处理一个实时指标的核心示例为了保持篇幅简洁省略部分细节Component public class MetricComputeConsumer { KafkaListener(topics ${metric.input.topic}, containerFactory kafkaContainerFactory) public void onMessage(ConsumerRecordString, String record) { MetricEnvelope envelope JsonUtil.parse(record.value(), MetricEnvelope.class); MetricConfig config metricRegistry.getConfig(envelope.getMetricType()); if (config null) { deadLetterService.submit(record); return; } boolean match expressionEngine.evaluate(config.getFilterExpression(), envelope.getPayload()); if (!match) { return; } Object result aggregatorChain.execute(config.getAggregatorChain(), envelope.getPayload()); resultStore.saveWindowedResult(config.getMetricKey(), result); notificationGateway.pushIfSubscribed(config.getMetricKey(), result); } }我不建议把代码写得过于抽象但上面这个核心链路的抽象程度是合适的每个组件只负责一件小事新指标上线时基本不用改 Java 代码只改配置。这带来的直接收益就是业务方提一个指标需求从提报到上线的时间从原来的一周压缩到了几小时。3.4 规则引擎的实现思路规则引擎我选择了 Groovy 脚本加一个受限的沙箱。每个规则脚本被严格限制执行时间默认500毫秒内运行在独立的线程池里如果超时会被强制中断。这样做确实有安全上的考量——不允许规则脚本访问系统类、文件、网络等敏感资源表达式语法上限制掉 System、Runtime、ProcessBuilder 等危险类。在金融内部业务场景这个安全级别完全够用。规则入参的设计也值得一提。每个规则会收到一个统一的上下文对象RuleContext里面包含当前指标的最新值、历史基线、同比环比、关联对象ID等。脚本里只需要写类似这样的代码if (ctx.currentValue ctx.baselineLowerBound * 0.8) { return new RuleAction(warning, 指标低于基线下限20%) } return new RuleAction(none, )这个上下文对象的设计让规则编写者完全不需要关注数据怎么查、怎么算只需要专注于业务判断逻辑大大降低了使用门槛。甚至业务分析师经过简单培训后也能自己写基础规则这就是平台类的价值所在。规则执行结果的输出也做了标准化。输出的RuleAction中包含动作类型和描述后续会交给动作执行器去处理。动作执行器实现了多种通道Webhook 回调、内部 IM 推送、短信发送、邮件通知。每种通道在实现时都做了限流和降级单个通道异常时不会影响其他通道这是预警系统在生产环境里必须具备的韧性。3.5 可观测性与监控设计一个数据平台如果自身没有监控那出了问题排查起来真的跟大海捞针一样。我在系统里埋了三类监控指标接入层监控上报量、校验失败率、Kafka 积压量、计算层监控处理延迟、聚合耗时、窗口命中率、服务层监控查询QPS、推送成功率、WebSocket 连接数。这些监控指标通过 Prometheus 采集在 Grafana 上展示。核心告警项包括Kafka 消费积压超过阈值、计算引擎任务执行超时率偏高、推送通道成功率低于99%。还有一个小细节是链路追踪。因为消息从接入到计算到推送要经过多个服务没有唯一追踪ID很难定位问题。我在消息进入接入层时生成了一个全局 traceId随消息一起写入 Kafka 的 header整个链路全程透传。日志框架里统一打印这个 traceId这样遇到“某条数据没算出来”的问题可以直接用 traceId 查所有日志节点效率提升不止一点半点。4. 我踩过的坑与调优记录4.1 消费端不均衡导致的计算倾斜平台上线初期我发现有个 Topic 的消费速度一直跟不上生产速度但看消费者组的整体积压量又不高。后来排查发现是 key 分布不均匀造成的某个热点业务方的数据占了总数据量的 70%消息全部 hash 到了同一个分区那个分区的消费者忙成狗其他消费者闲得长草。这个问题给了我一个很深的教训在选择 Kafka 分区 key 时不能直接用业务方ID而要结合消息内的一个子维度做组合散列。比如业务方ID加渠道ID必要时要再加一个随机盐确保消息能相对均匀地分散到各分区。修改之后分区消费量从“三七开”变得基本均衡。这属于那种你不在生产环境大规模跑一次就绝对发现不了的坑。4.2 Redis 大Key问题差点把缓存打爆做滑动窗口指标时最初把所有明细消息都塞进了同一个 ZSET导致某些热点指标对应的 ZSET 里数据量巨大单个 Key 的 value 达到了好几十MB。那次 Redis 的 INFO 里内存直接告警ZSET 操作延迟也飙到了几百毫秒连带着其他指标的计算也被拖慢了。后来做了两件事解决。第一引入“小窗口预聚合”秒级聚合成一条计数再进入 ZSET明细数据落 MySQL 做离线查询。第二给 ZSET 设置一个不过期的滚动清理策略每次写入时顺带删除超过窗口范围两倍的老数据避免数据只增不减。这个改动上线后Redis 内存占用下降了 70%ZSET 操作耗时重回亚毫秒级别。遇到这个问题的同时我也更加确信像 Redis 这类组件用得好不好很多时候取决于你对数据生命周期的设计而不是组件本身的性能。4.3 指标修正导致的口径争议前面提到我做了窗口修正策略但这个策略上线后引来了新问题业务方在午间看到“今日流水总额”出现了回退后台就炸了觉得数据算错了。其实就是先初算的版本少算了几条迟到的消息修正后金额反而更准确但业务方只看到了“数字变小”。为这件事开的会让我深刻意识到技术上的正确不一定等于业务上的可接受。后来的做法是对于对外展示的核心业务指标默认不启用修正覆盖而是通过一个“修订标记”来提示业务方当前结果可能会修正。同时在管理后台提供“修正生效时间”配置业务方如果追求极致的准确可以把允许延迟时间拉长前提是接受展示延迟。这些都是系统设计时容易忽略的“非功能需求”但实际运营中它们往往比功能需求更能决定项目的成败。你辛辛苦苦实现了高性能的计算引擎却因为在数据展示口径上没沟通好最后业务方还是觉得平台“不靠谱”那就太可惜了。4.4 预警风暴压测时才暴露预警模块在功能测试阶段没有发现问题因为测试数据量小触发频率低。但一次全链路压测中模拟了几倍于正常流量的数据预警服务中心瞬间打了几百条推送导致内部网关超时。当时我的第一反应不是调大推送线程池而是立刻想着怎么从源头控制预警数量。前面所说的预热、冷却、聚合三层策略正是在这种背景下实现的。压测结果验证了策略的有效性同等流量下最终发出的预警数量下降了一个数量级。这条经验也适用于所有事件驱动型系统预警机制必须有自我保护能力否则系统一但进入故障状态海量冗余通知会让故障雪上加霜。5. 关键配置与参数参考5.1 Kafka相关参数配置Kafka 参数这个东西很多人喜欢照抄网上的推荐值但不同业务场景差别真的很大。我这里给出的是消息体较小、高峰流量较大的金融数据场景下的一组参考配置。参数推荐值说明batch.size64KB批量发送阈值调大能提升吞吐linger.ms10ms等待更多消息的时间兼顾延迟和吞吐acksall确保写入副本成功compression.typelz4减少网络带宽占用max.poll.records500单次消费拉取条数避免堆积太多内存enable.auto.commitfalse手动提交偏移量防止重复消费或丢失关于enable.auto.commit我想多说一句生产环境建议一定关闭自动提交改用手动提交配合处理成功的业务逻辑。虽然手动提交代码上麻烦一点但能有效避免“消费了消息但处理失败却把偏移量提交了”的问题。很多数据不一致的隐形故障根源就出在这里。5.2 Redis与MySQL的存储划分这一块我做一个总体的划分。实时性要求高的指标结果、滑动窗口数据、去重集合全部放 Redis冷数据、历史明细、预警记录、配置快照放 MySQL。Redis 里我建议用 HASH 结构存指标快照用一个 KEY 表示“指标名”field 表示“统计周期”value 存 JSON 序列化的结果对象。这样查询一个指标多个时间段的数据时只需要一次 HGETALL非常高效。线下数据表的设计其实并没有很复杂但我在建表时做了一件降低后续心智负担的事情所有核心表都带一个record_snapshot_no字段每次全局指标快照生成时这个编号单调递增。业务方如果需要分析某一时刻的全局口径只需要通过快照编号去查询不需要跨表核对时间戳。5.3 告警配置参数模板告警配置我提供一个基础模板你可以根据项目情况微调。核心思路是告警不是越多越好要让预警真正能落地处理宁可少而准不可多而滥。alert: preheat_count: 3 # 连续触发多少次后才告警 cooldown_seconds: 300 # 同一规则告警冷却时间 aggregate_window: 60 # 聚合窗口秒 max_push_rate: 5 # 每秒钟最大推送条数 dynamic_baseline: enabled: true history_days: 7 lower_ratio: 0.8顺带提一下max_push_rate这个参数的意义。它不是限制业务方的接收能力而是保护平台自身。当出现大规模故障时限流会让预警延迟一点点时间但这远比把网关打爆导致所有通知都发不出去要靠谱得多。6. 后续扩展与个人体会这套平台上线运行之后业务方的反馈整体是积极的。过去一个新指标从提需求到看到数据动辄两三周现在基本当天就能在平台上配置上线。早期最核心的“指标口径不统一”问题也通过快照机制和统一配置中心得到了很大改善。我自己的体会是做这类平台型项目技术能力其实只是基础最难的部分往往是对业务的抽象以及和各种业务方在口径问题上的耐心沟通。如果后续要继续演进我比较看好的方向是引入一套轻量级的“指标血缘”能力。也就是说让每个指标能从数据源一路追溯到最终的报表展示中间经过了哪些过滤、哪些算子、哪些修正全部可视化呈现。当业务指标出现异常时顺着血缘图就能快速定位到是哪条链路的数据出了变化。这个能力目前市面上有一些商业产品在提供但自研实现一个基础版本并没有想象中那么难。最后再分享一个个人经验开发这类数据平台时一定要把“数据质量”当成一等公民在架构设计阶段就预留好校验、修复、对账的能力而不是等项目上线了再补。数据质量的基础设施建得越早后期业务扩张时你越从容。很多问题看着是业务规则问题深挖到底都是数据质量问题。地基稳了上面盖什么样的业务大楼都不怕。
前阵子一直在忙一个代号叫 financial-services 内部项目的重构这个项目说白了就是一套面向金融服务场景的通用数字化服务平台。不是那种给人一种高深莫测感觉的量化交易系统也不涉及任何复杂金融模型它更像是一套“运营中台”——负责把分散在不同业务方的行情数据、用户行为数据、账务流水数据统一收进来清洗、计算、存储再通过规则引擎和接口服务输出给前端展示、风险识别和业务决策使用。我之所以想把这次实践经验完整写出来是因为这个项目踩了不少坑也沉淀了一些通用方法论。不管你是做后端开发、数据工程还是刚到金融类公司做技术平台多多少少都能从中找到一些参考。尤其是一些架构选型的取舍、实时指标计算的细节以及生产环境里常见的脏数据和延迟问题这些都不是参考文档能告诉你的。如果你正准备搭一套类似的内部数据服务平台那这篇文章应该能帮你少走几条弯路。1. 项目背景与整体设计思路1.1 这个项目到底要解决什么问题先说背景。业务侧有多个渠道包括移动端、网页端、合作方接口都会产生大量的服务数据。过去这些数据各自为政业务A关心交易量业务B关心用户活跃度业务C关心异常波动每个团队各自拉数据、各自建表、各自开发报表最后口径对不上指标算出来的结果经常相差百分之二三十领导问起来谁也说不太清楚。financial-services 这个项目最初的目标非常朴素把数据统一收口建立一个稳定的指标计算平台。但做着做着发现光“统一收口”还不够因为业务方对实时性的要求越来越高。以前跑T1的离线报表还行现在业务方希望一个活动上线后能实时看到用户参与情况、交易成功率、异常波动预警。这就把项目的技术重心从“批量处理”推向了“实时/准实时计算”。所以最终这个项目的核心定位是三块数据接入标准化、指标计算统一化、预警输出自动化。听起来不复杂但真正落地时每一块都有不少细节。1.2 技术选型的思考过程选型永远不是选最火的而是选最适合团队维护的。因为初期调研的时候我也考虑过直接上完整的流式计算框架比如 Flink甚至也想过用 ClickHouse 做实时数仓。但最后综合团队情况选了一条更务实的路线Kafka 作为消息管道统一接入各类数据源一套自研的指标计算服务负责做清洗、聚合和规则判断结果写入 MySQL 和 Redis对外提供 REST API 和 WebSocket 推送。为什么没有用 Flink其实也不是不能而是学习成本和运维成本的问题。团队当时对 Java 和 Spring Boot 非常熟悉但对 Flink 的部署、调优、状态后端管理并不熟练。对于这样一个不算超大规模的指标计算场景用 Kafka 解耦自研聚合服务完全能支撑业务量而且出现问题时团队自己能快速排查。我个人的原则是在满足性能指标的前提下尽量选择团队有掌控力的技术栈不要为了简历好看去引入后续没人维护得动的东西。当然这套设计也有明显的短板。自研计算服务的水平扩展性不如 Flink 这类专业框架如果未来每秒写入量达到几十万条就不得不重构。但考虑到业务增长曲线这个风险在接下来两年内是可控的而且我也在设计上做了分片处理留了一条扩展的后路。1.3 架构的整体分层一个容易被忽略的点是架构图看着简单真正落地时要区分清楚数据分层和业务分层。数据层面我分了四层采集层、传输层、计算层、服务层。采集层由各业务线通过 SDK 或 API 上报数据传输层是 Kafka负责削峰填谷计算层是核心承载数据清洗、指标计算、规则匹配服务层对外提供查询接口和消息推送。业务层面则是围绕“指标”来组织而不是围绕“数据源”来组织。比如用户活跃度这个指标它可能来源于用户埋点、登录日志、接口调用日志等多个数据源。如果按数据源组织逻辑后期指标扩展时会非常痛苦按指标组织逻辑新增数据源只需要在接入层做一次适配后续所有指标都能自动复用。这个设计看起来是架构层面的小事但其实决定了后续所有开发效率。我还做了一个很关键的决策计算服务内部分了两条链路一条是实时链路一条是准实时链路。实时链路处理延迟敏感类的指标比如今日交易总额、当前在线用户数秒级刷新准实时链路处理需要跨时段对比的指标比如环比增长率、异常波动检测分钟级计算。这样做的好处是不会因为少数复杂指标拖慢整体实时性也让开发时可以按需取舍计算深度。2. 核心模块拆分与细节解析2.1 数据接入层管好格式才能少踩坑数据接入是整个平台的地基也是最容易出现脏数据的地方。接入层我采用了 Topic 加版本号的策略而不是简单一个 Topic 走天下。每个业务方上报数据时必须带上 schema 版本号没有带的一律进“待修复队列”通过一个管理后台让业务方自己去查看和修正。前期会有些阵痛业务方觉得怎么这么麻烦但到了后期做指标对齐时你就会发现这个约束实在救了大命。协议上当时也纠结过是用 JSON 还是 Protobuf。坦诚讲JSON 在调试时非常友好但做复杂嵌套结构的时候校验成本太高解析性能也一般。最终用了 JSON JSON Schema 校验而不是直接上 Protobuf。原因是团队当时人手不足Protobuf 的代码生成和版本管理需要额外投入而 JSON Schema 允许业务方在管理后台通过配置来调整校验规则灵活度高很多。等业务量上来之后再考虑迁移到 Protobuf也是一个可行的演进路径。接入层的另一个隐藏细节是“幂等”。Kafka 提供了至少一次投递语义这在数据上报场景下意味着重复消息几乎一定会出现。我们在接入层生成了一个针对消息体内容的 hash 值存在 Redis 里做去重过期时间设置为 24 小时。这个处理看似简单但极大地减少了后续计算层的重复计数问题尤其是做交易量这类累加型指标时没有幂等处理数据会有很多水分。2.2 指标计算层窗口、口径与聚合一锅端指标计算层是平台的大脑也是代码量最集中的地方。我在这一层实现了一套轻量级的“指标引擎”核心思路是声明式配置每个指标由三个要素定义——数据源过滤条件聚合逻辑。数据源决定去哪些 Topic 取数过滤条件决定哪些记录参与计算聚合逻辑负责怎么算。业务方新增一个指标时只要走配置中心提交一份定义计算服务会自动编译成可执行的算子链不需要重新发版。可能有人会问为什么不用现成的流计算平台或者直接用 SQL 去定义指标其实在前期调研时SQL确实是备选方案但实际业务中有很多非标准化的逻辑例如“剔除内部测试账号”“排除特定来源的重复请求”这些用SQL表达比较绕。把指标计算留在代码里用配置驱动反而最灵活。窗口计算是这个模块里最难啃的骨头。实时指标往往涉及到滑动窗口比如“过去5分钟的交易成功率”。要实现这个我用 Redis 的 ZSET 按时间戳存储每个数据点每次计算时通过 score 范围查询取出窗口内的数据再批量执行聚合。这个方案在数据量不大时性能很好看但一旦窗口内数据量突破几十万ZSET 的操作延迟就会明显上升。后来我做了采样策略对于高吞吐的指标在写入 Redis 之前做一次预聚合把单条消息聚合成秒级的计数再把秒级计数存入 ZSET。这个优化直接将窗口查询的数据量减少了两个数量级。聚合一端还需要特别注意“重算”的问题。消息延迟乱序在分布式场景下是常态比如一条 10:00:01 的数据可能 10:01:30 才到。如果你在窗口关闭的瞬间就固化结果这条迟到的数据就永久丢失了。我对每个指标配置了一个“允许延迟时间”默认 30 秒。窗口关闭后先出一版结果延迟时间到期后再做一次修订计算如果结果有变化推送一条修订消息给下游。这种“先初算、后修正”的模式在业务容忍范围内是一个极其实用的折中方案。2.3 规则引擎与预警中心让异常自动被发现业务方对平台的一个核心期望是“别让我们天天盯着报表异常了你得告诉我们”。这就需要一个简单但高效的规则引擎。我这里没有引入 Drools 这类重量级规则框架而是实现了一个基于 Groovy 脚本的轻量级规则沙箱。每条规则由两部分组成触发条件一段表达式动作推送消息、打标记、调用回调接口。规则配置支持灰度发布可以先让 10% 的流量实验观察效果没问题再全量开放。规则容易写但预警风暴难防。最常见的情况是某个指标在短时间内抖动规则频繁触发业务方收到几十条同一个原因的告警慢慢就会产生“告警疲劳”把推送直接屏蔽掉。踩过这个坑之后我特意实现了在不同规则上的灵活防护策略一是预热规则触发后要连续 N 次持续命中才真正发出预警二是冷却同一规则两次预警之间必须至少间隔 5 分钟三是聚合同一对象的多条预警合并成一条摘要摘要里附带最近一段时间的趋势数据。这样调整后的实际效果告警量减少了约 60%业务方反而更愿意处理预警信息了。规则引擎里还有一个容易被忽略的能力动态阈值。有些指标在业务高峰和低谷时的波动差异很大如果阈值写死不是误报就是漏报。我的方案是让平台自动维护每个指标的历史置信区间以过去7天同一时间段的均值加减三倍标准差作为动态基线规则配置时可以引用这个基线。比如可以设定“当指标值低于基线下限的20%时触发预警”这个能力上线之后异常识别的准确率提升明显。2.4 对外服务层既要查得到也要推得动平台对外提供的服务主要有两类查询类 API 和实时推送通道。查询类 API 相对常规基于 MySQL 和 Redis 的一二级缓存模式热点指标走 Redis历史明细走 MySQL。这里有一个经验值得分享对于指标查询接口我强烈建议不要在 API 层再引入复杂的OR映射框架直接用一个轻量的查询对象组装 SQL 即可。因为指标查询基本都是固定模式的聚合查询过度设计只会增加排查问题的难度。实时推送这一块用的是 WebSocket配合一个轻量的推送网关。网关维护着每个业务方的订阅列表当指标计算结果有更新时把变更记录推给所有订阅了该指标的服务端连接。为了不让推送消息成为瓶颈推送频率做了限流同一指标每秒最多推送一条合并后的快照数据。实际运营下来推送最考验的不是吞吐而是连接状态的维护——客户端断线重连、心跳超时这些看似简单的问题在实际联调中比写核心逻辑还磨人。查询 API 还有一个磨人的点就是“指标对齐”。同样的“今日总交易额”不同端展示不一样会让人对平台失去信任。所以我在服务层做了一个快照表每五分钟生成一次所有核心指标的全局快照所有查询默认读取快照而不是实时精确值。这样不同服务方看到的数据是一致的只是有最多五分钟的延迟。对于对实时性要求更高的看板提供实时接口但会明确标注是“实时值”避免混用。3. 实操过程从0到1的实现路径3.1 整体技术栈与项目结构整个项目后端使用 Java 17 Spring Boot 3.x消息中间件用 Kafka 3.x缓存用 Redis 6.x存储用 MySQL 8.x。布式环境那就不用想了刚开始是在单机环境开发调试之后通过 Docker-Compose 部署到三台测试服务器上。项目代码本身按 Maven 多模块组织大致结构如下financial-services ├── financial-common # 公共代码工具类、统一返回结构、异常定义 ├── financial-collect # 数据接入服务接收HTTP上报、校验、写Kafka ├── financial-compute # 指标计算服务消费Kafka、清洗聚合、写Redis/MySQL ├── financial-rule # 规则引擎服务加载规则、执行判定、触发动作 ├── financial-api # 查询与推送服务REST API WebSocket推送 └── financial-admin # 管理后台指标配置、数据修复、规则配置对于刚开始做类似平台的同学我建议即使项目很小也尽量按这种纵向切分来组织模块。因为接入、计算、规则、服务四块的部署频率完全不同混在一个单体应用里一次计算逻辑的改动可能还要连带推送服务重新发布非常容易引发线上事故。3.2 接入与校验链路的实现接入服务本质上是替业务方扛住所有“可能出问题”的地方。业务方只需要按约定把一个数据结构 POST 过来剩下的工作全部在 financial-collect 里完成。数据接入的代码核心分三步接收请求、schema 校验、投递到 Kafka。第一步比较简单直接用一个 Controller 接收 JSON 字符串不急着反序列化成强类型对象因为后续字段可能变化强类型反而限制灵活性。第二步是校验用预加载的 JSON Schema 对消息体做校验失败的进入修复队列并通知业务方。第三步是投递这里关键在于为每条消息生成业务主键式的去重ID把来源标识、业务ID、发生时间拼接后做 MD5。实现时想说一个前面提到的坑接入层在处理量突然翻倍时会遇到线程池打满导致请求超时的情况。常规思路是调大线程池但真正的问题是 Kafka 生产者发送后等待确认的时间过长。后来我在代码里把 batch.size 从默认的 16KB 调到了 64KBlinger.ms 调整到 10mssend 的吞吐量明显上涨CPU 反而下降了。这种调优不复杂但需要你在压测时关注生产者的各项指标曲线。3.3 指标计算的代码骨架计算服务的代码骨架是消费 Kafka 消息、按指标路由、执行过滤条件、执行聚合算子、写入结果。这个骨架本身不新奇但有几个细节我认为值得分享出来。第一是消息路由。我收到了业务方上报的消息并不直接按照消息类型去查表而是在消息的 header 里塞入了动态类型标记比如metric-type: user_active消费端拿到后直接按这个标记定位到预先注册的计算单元省去了解析嵌套JSON后再判断类型的损耗。第二是过滤条件的处理。过滤条件我实现成一个 Python 风格的表达式引擎引入一个轻量级表达式解析库配置人员可以在管理后台写类似channel ! internal_test and amount 0这样的表达式保存后会编译成一个可执行的对象每次消息到达时在内存中执行。这样既摆脱了硬编码又不会因为频繁拼接字符串而误伤性能。第三是聚合算子。聚合算子根据指标类型分成 Sum、Count、Avg、DistinctCount、MaxMin 五类。每一类算子都有一个对应的装饰链比如 Counter 算子会先经过去重器再到累加器再到窗口存储。这种装饰链模式的好处是新增一个指标类型时不需要改聚合核心只要在配置里指定算子链即可。下面是计算服务消费并处理一个实时指标的核心示例为了保持篇幅简洁省略部分细节Component public class MetricComputeConsumer { KafkaListener(topics ${metric.input.topic}, containerFactory kafkaContainerFactory) public void onMessage(ConsumerRecordString, String record) { MetricEnvelope envelope JsonUtil.parse(record.value(), MetricEnvelope.class); MetricConfig config metricRegistry.getConfig(envelope.getMetricType()); if (config null) { deadLetterService.submit(record); return; } boolean match expressionEngine.evaluate(config.getFilterExpression(), envelope.getPayload()); if (!match) { return; } Object result aggregatorChain.execute(config.getAggregatorChain(), envelope.getPayload()); resultStore.saveWindowedResult(config.getMetricKey(), result); notificationGateway.pushIfSubscribed(config.getMetricKey(), result); } }我不建议把代码写得过于抽象但上面这个核心链路的抽象程度是合适的每个组件只负责一件小事新指标上线时基本不用改 Java 代码只改配置。这带来的直接收益就是业务方提一个指标需求从提报到上线的时间从原来的一周压缩到了几小时。3.4 规则引擎的实现思路规则引擎我选择了 Groovy 脚本加一个受限的沙箱。每个规则脚本被严格限制执行时间默认500毫秒内运行在独立的线程池里如果超时会被强制中断。这样做确实有安全上的考量——不允许规则脚本访问系统类、文件、网络等敏感资源表达式语法上限制掉 System、Runtime、ProcessBuilder 等危险类。在金融内部业务场景这个安全级别完全够用。规则入参的设计也值得一提。每个规则会收到一个统一的上下文对象RuleContext里面包含当前指标的最新值、历史基线、同比环比、关联对象ID等。脚本里只需要写类似这样的代码if (ctx.currentValue ctx.baselineLowerBound * 0.8) { return new RuleAction(warning, 指标低于基线下限20%) } return new RuleAction(none, )这个上下文对象的设计让规则编写者完全不需要关注数据怎么查、怎么算只需要专注于业务判断逻辑大大降低了使用门槛。甚至业务分析师经过简单培训后也能自己写基础规则这就是平台类的价值所在。规则执行结果的输出也做了标准化。输出的RuleAction中包含动作类型和描述后续会交给动作执行器去处理。动作执行器实现了多种通道Webhook 回调、内部 IM 推送、短信发送、邮件通知。每种通道在实现时都做了限流和降级单个通道异常时不会影响其他通道这是预警系统在生产环境里必须具备的韧性。3.5 可观测性与监控设计一个数据平台如果自身没有监控那出了问题排查起来真的跟大海捞针一样。我在系统里埋了三类监控指标接入层监控上报量、校验失败率、Kafka 积压量、计算层监控处理延迟、聚合耗时、窗口命中率、服务层监控查询QPS、推送成功率、WebSocket 连接数。这些监控指标通过 Prometheus 采集在 Grafana 上展示。核心告警项包括Kafka 消费积压超过阈值、计算引擎任务执行超时率偏高、推送通道成功率低于99%。还有一个小细节是链路追踪。因为消息从接入到计算到推送要经过多个服务没有唯一追踪ID很难定位问题。我在消息进入接入层时生成了一个全局 traceId随消息一起写入 Kafka 的 header整个链路全程透传。日志框架里统一打印这个 traceId这样遇到“某条数据没算出来”的问题可以直接用 traceId 查所有日志节点效率提升不止一点半点。4. 我踩过的坑与调优记录4.1 消费端不均衡导致的计算倾斜平台上线初期我发现有个 Topic 的消费速度一直跟不上生产速度但看消费者组的整体积压量又不高。后来排查发现是 key 分布不均匀造成的某个热点业务方的数据占了总数据量的 70%消息全部 hash 到了同一个分区那个分区的消费者忙成狗其他消费者闲得长草。这个问题给了我一个很深的教训在选择 Kafka 分区 key 时不能直接用业务方ID而要结合消息内的一个子维度做组合散列。比如业务方ID加渠道ID必要时要再加一个随机盐确保消息能相对均匀地分散到各分区。修改之后分区消费量从“三七开”变得基本均衡。这属于那种你不在生产环境大规模跑一次就绝对发现不了的坑。4.2 Redis 大Key问题差点把缓存打爆做滑动窗口指标时最初把所有明细消息都塞进了同一个 ZSET导致某些热点指标对应的 ZSET 里数据量巨大单个 Key 的 value 达到了好几十MB。那次 Redis 的 INFO 里内存直接告警ZSET 操作延迟也飙到了几百毫秒连带着其他指标的计算也被拖慢了。后来做了两件事解决。第一引入“小窗口预聚合”秒级聚合成一条计数再进入 ZSET明细数据落 MySQL 做离线查询。第二给 ZSET 设置一个不过期的滚动清理策略每次写入时顺带删除超过窗口范围两倍的老数据避免数据只增不减。这个改动上线后Redis 内存占用下降了 70%ZSET 操作耗时重回亚毫秒级别。遇到这个问题的同时我也更加确信像 Redis 这类组件用得好不好很多时候取决于你对数据生命周期的设计而不是组件本身的性能。4.3 指标修正导致的口径争议前面提到我做了窗口修正策略但这个策略上线后引来了新问题业务方在午间看到“今日流水总额”出现了回退后台就炸了觉得数据算错了。其实就是先初算的版本少算了几条迟到的消息修正后金额反而更准确但业务方只看到了“数字变小”。为这件事开的会让我深刻意识到技术上的正确不一定等于业务上的可接受。后来的做法是对于对外展示的核心业务指标默认不启用修正覆盖而是通过一个“修订标记”来提示业务方当前结果可能会修正。同时在管理后台提供“修正生效时间”配置业务方如果追求极致的准确可以把允许延迟时间拉长前提是接受展示延迟。这些都是系统设计时容易忽略的“非功能需求”但实际运营中它们往往比功能需求更能决定项目的成败。你辛辛苦苦实现了高性能的计算引擎却因为在数据展示口径上没沟通好最后业务方还是觉得平台“不靠谱”那就太可惜了。4.4 预警风暴压测时才暴露预警模块在功能测试阶段没有发现问题因为测试数据量小触发频率低。但一次全链路压测中模拟了几倍于正常流量的数据预警服务中心瞬间打了几百条推送导致内部网关超时。当时我的第一反应不是调大推送线程池而是立刻想着怎么从源头控制预警数量。前面所说的预热、冷却、聚合三层策略正是在这种背景下实现的。压测结果验证了策略的有效性同等流量下最终发出的预警数量下降了一个数量级。这条经验也适用于所有事件驱动型系统预警机制必须有自我保护能力否则系统一但进入故障状态海量冗余通知会让故障雪上加霜。5. 关键配置与参数参考5.1 Kafka相关参数配置Kafka 参数这个东西很多人喜欢照抄网上的推荐值但不同业务场景差别真的很大。我这里给出的是消息体较小、高峰流量较大的金融数据场景下的一组参考配置。参数推荐值说明batch.size64KB批量发送阈值调大能提升吞吐linger.ms10ms等待更多消息的时间兼顾延迟和吞吐acksall确保写入副本成功compression.typelz4减少网络带宽占用max.poll.records500单次消费拉取条数避免堆积太多内存enable.auto.commitfalse手动提交偏移量防止重复消费或丢失关于enable.auto.commit我想多说一句生产环境建议一定关闭自动提交改用手动提交配合处理成功的业务逻辑。虽然手动提交代码上麻烦一点但能有效避免“消费了消息但处理失败却把偏移量提交了”的问题。很多数据不一致的隐形故障根源就出在这里。5.2 Redis与MySQL的存储划分这一块我做一个总体的划分。实时性要求高的指标结果、滑动窗口数据、去重集合全部放 Redis冷数据、历史明细、预警记录、配置快照放 MySQL。Redis 里我建议用 HASH 结构存指标快照用一个 KEY 表示“指标名”field 表示“统计周期”value 存 JSON 序列化的结果对象。这样查询一个指标多个时间段的数据时只需要一次 HGETALL非常高效。线下数据表的设计其实并没有很复杂但我在建表时做了一件降低后续心智负担的事情所有核心表都带一个record_snapshot_no字段每次全局指标快照生成时这个编号单调递增。业务方如果需要分析某一时刻的全局口径只需要通过快照编号去查询不需要跨表核对时间戳。5.3 告警配置参数模板告警配置我提供一个基础模板你可以根据项目情况微调。核心思路是告警不是越多越好要让预警真正能落地处理宁可少而准不可多而滥。alert: preheat_count: 3 # 连续触发多少次后才告警 cooldown_seconds: 300 # 同一规则告警冷却时间 aggregate_window: 60 # 聚合窗口秒 max_push_rate: 5 # 每秒钟最大推送条数 dynamic_baseline: enabled: true history_days: 7 lower_ratio: 0.8顺带提一下max_push_rate这个参数的意义。它不是限制业务方的接收能力而是保护平台自身。当出现大规模故障时限流会让预警延迟一点点时间但这远比把网关打爆导致所有通知都发不出去要靠谱得多。6. 后续扩展与个人体会这套平台上线运行之后业务方的反馈整体是积极的。过去一个新指标从提需求到看到数据动辄两三周现在基本当天就能在平台上配置上线。早期最核心的“指标口径不统一”问题也通过快照机制和统一配置中心得到了很大改善。我自己的体会是做这类平台型项目技术能力其实只是基础最难的部分往往是对业务的抽象以及和各种业务方在口径问题上的耐心沟通。如果后续要继续演进我比较看好的方向是引入一套轻量级的“指标血缘”能力。也就是说让每个指标能从数据源一路追溯到最终的报表展示中间经过了哪些过滤、哪些算子、哪些修正全部可视化呈现。当业务指标出现异常时顺着血缘图就能快速定位到是哪条链路的数据出了变化。这个能力目前市面上有一些商业产品在提供但自研实现一个基础版本并没有想象中那么难。最后再分享一个个人经验开发这类数据平台时一定要把“数据质量”当成一等公民在架构设计阶段就预留好校验、修复、对账的能力而不是等项目上线了再补。数据质量的基础设施建得越早后期业务扩张时你越从容。很多问题看着是业务规则问题深挖到底都是数据质量问题。地基稳了上面盖什么样的业务大楼都不怕。
