★ FEATURED ARTICLE 接手第一个内部工具项目的时候我给它起名叫 dig1.0全称是 Digital Integration Gateway 1.0也就是“数据集成网关”的第一个可落地版本。当时团队里没有专职的数据平台组后端同学各自负责一个微服务系统之间要同步订单、会员、库存这些数据最原始的办法就是写一堆定时任务你调我的接口、我读你的表时间一长回调超时、字段对不上、重复导入的问题全冒出来了。所以我决定做一个统一的数据接入和分发工具把“从哪里拿数据、怎么处理数据、送到哪里去”这三件事从业务代码里抽出来集中管理。这篇文章就把 dig1.0 从需求拆解到落地实现的完整过程记录下来里面包含了设计思路、核心代码骨架、部署验证方式还有我在实际运行中踩过的坑。如果你也在维护系统间数据同步或者在考虑自建一套轻量的数据接入层这篇内容可以直接拿去做参考。1. 需求拆解与整体设计思路1.1 第一版要解决什么问题在动手写代码之前我先把“痛点”列成了清单不是拍脑袋决定功能而是跟着数据流的路径来梳理。当时的典型场景有三个第一订单系统每天要定时把增量订单推给财务系统第二CRM 系统要同步会员表里的基础信息第三需要把第三方开放接口返回的数据拉回来经过清洗后写入本地数据库。这三类场景看起来不同本质却有共性都需要一个“源”把数据从某个存储或接口里读出来都需要一个“处理过程”做字段映射、格式转换、空值过滤都需要一个“目标”把处理后的数据写进某个表或推给某个接口。所以我给 dig1.0 定义的目标不是做一个万能的 ETL而是做一个轻量的数据管道支持“定时轮询 处理链 多目标分发”。我不会一上来就引入大数据组件因为团队规模摆在那里没有专职运维也没有大数据集群用一个能打成 Jar 包部署的 Spring Boot 应用比 Flink、Canal 这类重型方案轻得多。轻意味着好维护出问题容易回滚业务同学也不需要学一套新的开发范式。1.2 技术选型背后的取舍技术选型没有绝对的对只有合不合适。dig1.0 最终选择了 Java 8 Spring Boot 2.x原因有三个团队现有代码大部分是 Java接入成本最低Spring Boot 自带配置管理和依赖注入适合做“插件式”扩展JVM 生态里的连接池、HTTP 客户端、日志组件都很成熟不需要自己造轮子。数据源方面数据库读取用了 JDBC 模板接口拉取用了 RestTemplate目标端写入继续用 JDBC 模板。内部流转的队列没有引入消息中间件而是用了 JDK 自带的 BlockingQueue。当时也有同事建议上 RocketMQ但考虑到第一版的数据量每天也就几十万条单机内存队列足够应付而且少一个中间件就少一个故障点。如果未来数量级上来再替换成消息队列也不迟接口层统一抽象了替换成本可控。我对 dig1.0 的定位是“把简单的事做扎实”。处理链不做得花哨但是每一步都必须有日志重试机制不做得琐碎但是必须保证数据不丢并发不一定要多高但是要能通过配置调整而不是写死。1.3 命名与版本号的含义“dig”不是挖矿的意思在这个项目里是 Digital Integration Gateway 的缩写。起这个名字就是希望它像一个网关流量从入口进来经过处理从出口出去。1.0 表示这是第一个正式交付的版本意味着可以承担内部业务但不是终点。版本号定得很保守因为我知道第一版一定会有各种问题。为了不给自己留退路我在设计时就约定代码里所有输入输出必须带 requestId请求标识处理链的每一步都要记录耗时Sink 写入必须支持幂等。这些约定不复杂但是能在出问题时快速定位也让我在版本迭代时心里有底。2. 核心模块与关键实现2.1 数据源适配层dig1.0 的第一个核心模块是 Source也就是数据源适配层。我把它设计成一个接口任何数据源只需要实现四个方法初始化、拉取数据、确认完成、关闭资源。定义代码如下public interface Source { void init(SourceConfig config); ListDataRecord poll(int batchSize); void ack(ListDataRecord records); void close(); }为什么要单独拆出 ack 方法因为在真实场景里拉取数据和确认数据不是同一时刻发生的。比如从数据库表轮询增量数据我拉取了 id 大于某个位点的二十条记录在把这二十条记录放入处理队列后必须等管道处理成功才能把位点更新到最新。如果刚拉出来就更新位点一旦应用在写入前崩溃这批数据就丢了。我把 ack 设计成“批量确认”处理成功的记录集合会传回给 Source由具体的 Source 实现类决定怎么记录位点。对于数据库表我推荐用一张 offset 表存储各表的同步位点对于接口拉取则把游标和下一页标识存到本地文件或数据库。这样做的好处是重启之后还能接着跑不会全量重扫。2.2 管道处理链数据从 Source 拉出来后会进入处理链 Pipeline。我没有选择写一个超大的 Processor 类来承载所有逻辑而是把处理动作拆成三个基本节点Filter 负责过滤Transform 负责转换Router 负责路由。每一批 DataRecord 会依次经过这三个节点。Filter 做的是最无脑的过滤比如剔除状态为“已删除”的记录、过滤掉时间戳超过七天的数据。Transform 负责字段映射和格式转换比如把数据库里的 createTime 从时间戳转成字符串把用户性别从 0/1 转成“男/女”把手机号做脱敏。Router 则根据规则决定这条记录应该投递到哪个 Sink 目标。这里有一个容易忽略的点处理链里的每个节点都必须是无状态的。所谓无状态是指节点不持有自己的可变成员变量所有临时数据都放在 DataRecord 的上下文里。因为管道里的数据是批量并发的如果有共享状态就得处理并发控制无状态化之后每个节点都像纯函数一样传入一条记录返回处理后的结果天然线程安全。2.3 分发器与目标端管道处理完一条数据后Router 会拿到目标标识然后交给 Sink。Sink 同样是接口化设计目标是 MySQL 就实现一个 MySQLSink目标是 HTTP 接口就实现一个 HttpSink目标是消息队列就实现一个 MqSink。这样业务方要接入新目标只需要新增一个 Sink 实现类不需要改动管道逻辑。Sink 写入时我特别强调幂等。以 MySQL 写入为例不是简单执行一条 insert而是依赖唯一键做 upsertINSERT INTO target_table (id, name, update_time) VALUES (?, ?, ?) ON DUPLICATE KEY UPDATE name VALUES(name), update_time VALUES(update_time);这样无论同一条数据被处理多少遍最终落到库里的结果都是一致且唯一的。对于 HTTP 类型的 Sink幂等则依赖请求头里的 requestId让接收方通过 requestId 去重。这个设计在后来的运行里救了不少次因为数据重复消费几乎是必然会遇到的。3. 实操过程与核心代码骨架3.1 项目目录结构与配置我在这里给出一个可以照抄的骨架项目名叫 dig1.0基于 Maven 构建主包名用 io.example.dig业务代码按下面的结构组织dig1.0 ├── pom.xml ├── src │ ├── main │ │ ├── java │ │ │ └── io/example/dig │ │ │ ├── DigApplication.java │ │ │ ├── core │ │ │ │ ├── Source.java │ │ │ │ ├── Sink.java │ │ │ │ ├── DataRecord.java │ │ │ │ └── Pipeline.java │ │ │ ├── source │ │ │ │ ├── JdbcSource.java │ │ │ │ └── HttpSource.java │ │ │ ├── sink │ │ │ │ ├── JdbcSink.java │ │ │ │ └── HttpSink.java │ │ │ └── task │ │ │ └── SyncTask.java │ │ └── resources │ │ └── application.ymlapplication.yml 里最关键的配置是每个任务的开关、数据源连接信息和目标连接信息。我会给每个任务定义一个简短的名称比如 order_to_finance后续所有日志、位点、重试记录都会带上这个任务名。dig: tasks: order_to_finance: enabled: true source: type: jdbc url: jdbc:mysql://localhost:3306/order_db username: dig_user password: change_me sink: type: jdbc url: jdbc:mysql://localhost:3306/finance_db username: dig_user password: change_me pollBatchSize: 100 pollIntervalMs: 5000配置尽量扁平化不要套太深的层级。当时为了让代码看起来“灵活”我一度想做成动态配置中心热更新后来发现第一版根本不需要配置文件加一个重启就能解决问题。过度设计会拖慢第一个版本的交付节奏。3.2 核心代码的骨架Source 实现类我写一个最简单的 JdbcSource它根据配置文件里的位点字段每次轮询都查询“更新时间大于同步位点”的数据。位点存放在一张叫 dig_offset 的表中key 是任务名value 是位点值。public class JdbcSource implements Source { private JdbcTemplate jdbcTemplate; private String querySql; private String offsetKey; Override public ListDataRecord poll(int batchSize) { long offset getOffset(offsetKey); ListMapString, Object rows jdbcTemplate.queryForList(querySql LIMIT ?, offset, batchSize); return rows.stream() .map(row - DataRecord.fromMap(row)) .collect(Collectors.toList()); } Override public void ack(ListDataRecord records) { if (records.isEmpty()) { return; } Date maxUpdateTime records.stream() .map(record - record.getField(update_time, Date.class)) .max(Date::compareTo) .orElse(null); if (maxUpdateTime ! null) { jdbcTemplate.update(UPDATE dig_offset SET offset_value ? WHERE offset_key ?, maxUpdateTime.getTime(), offsetKey); } } private long getOffset(String key) { Long value jdbcTemplate.queryForObject( SELECT offset_value FROM dig_offset WHERE offset_key ?, Long.class, key); return value null ? 1L : value; } }Pipeline 的处理逻辑不复杂核心就是按顺序调用每个节点。我会把 Filter、Transform、Router 都放在同一个 List 里借助 Spring 的注入机制按任务维度组装不同的处理链。public class Pipeline { private ListProcessor processors; public DataRecord process(DataRecord record) { for (Processor processor : processors) { long start System.currentTimeMillis(); DataRecord result processor.process(record); long cost System.currentTimeMillis() - start; log.info(requestId{}, processor{}, cost{}ms, record.getRequestId(), processor.name(), cost); if (result null) { return null; } record result; } return record; } }Sink 的代码骨架要注意批量提交和失败重试的配合。直接每条记录单独提交性能不好批量提交又会放大失败的影响面。我的策略是每批最多一百条如果批量提交失败就把这一批拆成单条逐一重试单条也失败的就进入死信队列等待人工处理。3.3 部署与验证步骤dig1.0 没有做复杂的安装流程就是打一个 Spring Boot 的可执行 Jar 包放到一台有 Java 环境的两核四G机器上跑。启动命令很简单java -Xms512m -Xmx512m -jar dig1.0.jar --spring.config.locationfile:/path/to/application.yml验证是否正常我会先用一个手工构造的任务来测。比如在源订单表里插入一条状态为“待同步”的订单号 TEST001然后观察日志里是否出现“taskorder_to_finance, statussuccess, requestIdxxx”的记录。接着查目标 finance 库的对应表确认 TEST001 已经出现。这里要特别提醒不要在联调环境里一上来就接全量真实数据。先用一条脏数据跑通链路确认 Source、Pipeline、Sink 三个环节都没有问题再开增量任务。我当时图省事直接连着生产只读库做首次联调结果因为字段映射写错把一批真实订单的金额字段搞空了虽然最后通过幂等重跑修回来了但整个过程很狼狈。4. 常见问题与排查实录4.1 数据重复消费下游出现重复记录这个问题是在接入 CRM 系统会员同步时暴露的。任务跑了一个小时会员表里出现了大量重复记录一开始以为是 Sink 写了两遍后来查日志发现是 Source 的 ack 时机不对。JdbcSource 在 poll 里查询的数据会先放进内存队列但还没有处理完管道中某一批记录因为网络抖动导致重试重试时又重新 poll 了一遍于是同一批数据被塞进队列两次。解决方式有两层。第一在 JdbcSource 拉取数据时把查询结果里的主键集合作为批次缓存在处理完成前不会再次拉取相同主键第二在 Sink 端坚持使用 upsert 幂等写入。这两层缺一不可Source 层的去重能减少无效计算Sink 层的幂等能兜底保证数据最终一致。4.2 处理性能瓶颈任务越跑越慢上线跑了一周后财务同步任务的平均处理耗时从 2 秒涨到了 20 秒。刚开始怀疑是服务器资源不够看了 CPU 和内存都很低。后来定位到是 Source 查询语句的问题增量查询条件用了 update_time offset但 update_time 字段没有建索引数据量一上来每次 poll 都是全表扫描数据库 CPU 被打满。解决办法不是先加服务器而是先把数据库索引加上。我给源表的 update_time 字段建了普通索引同时把查询 SQL 改成只查必要的字段不要 select *。索引建好后单次 poll 从 800 毫秒降到了 30 毫秒整条管道又恢复了正常。这个教训提醒我第一版上线前就应该检查所有查询条件的索引覆盖情况而不是等到性能报警了才回头排查。4.3 下游重试风暴接口被压垮和第三方接口对接时如果目标方响应超时HttpSink 会自动重试。我最初的做法是失败后立即重试每次间隔 1 秒最多重试 5 次。结果某个第三方接口持续抖动时dig1.0 的平均重试请求数每秒超过两百个反而把对方接口打得更不稳定最后对方反馈流量异常我们这边需要紧急降级。之后我改成了指数退避重试第一次失败后等待 2 秒第二次等 4 秒第三次等 8 秒最大间隔不超过 60 秒。同时引入死信队列同一个 requestId 连续失败 5 次之后不再自动重试而是落库通知人工处理。重试机制不是越激进越好要对下游保持礼貌。这里整理一个速查表方便后续排查同类问题问题现象可能原因排查思路解决方案下游出现重复记录ack 过早或重试导致重复 poll查日志确认同一 requestId 是否多次处理Source 批次缓存 Sink 幂等写入任务越来越慢查询字段无索引、select * 全字段扫描看数据库慢查询日志、EXPLAIN 执行计划补索引、精简查询字段接口超时后联锁反应固定间隔重试导致流量放大看重试日志和下游接收方监控指数退避、设置最大重试次数重启后任务从错误位点执行位点存储不完整或原子性差检查 offset 表更新是否与 Sink 成功绑定把 ack 放到 Sink 成功提交之后再执行5. 复盘与后续演进5.1 从 1.0 到 1.1 的规划dig1.0 解决了几个具体场景但也暴露了很多不足。先说配置管理当前所有任务配置都躺在 application.yml 里新增一个任务要改文件再重启这对交付效率是不小的损耗。后续我计划把配置迁移到配置中心做到新增数据源、修改处理链都能动态生效不用重启进程。任务调度也是 1.1 的重头戏。当前每个任务都通过内置的定时线程池触发任务少还好说任务多了之后线程池的管理和不同任务之间的优先级划分就会变得混乱。我会引入一个简单的调度器或者把任务注册表做成可以按优先级排序的模型让重要任务的轮询频率更高。数据源适配器也需要扩充。当前支持 JDBC 和 HTTP下一步打算支持从 CSV 文件导入、从消息队列订阅消费以及对象存储里的文件扫描。适配器变多之后Source 接口还需要增加一个“探测”能力比如启动时自动检测源端连通性避免某个源无效时整个任务运行期都在空转。5.2 落地使用的心得如果你也想在团队里做类似的数据接入网关我的建议是控制住第一版的范围不要想着做成一个数据中台。dig1.0 从一开始就只承诺三件事把数据拉过来、做基础处理、稳定送出去。像数据血缘、自适应清洗、可视化编排这些听起来很酷的能力通通放到后续版本里先把管道跑通、把日志和监控补上才是真正有价值的第一步。日志和监控绝对是第一版必须做好的事。我们在 dig1.0 里给每一批数据的每个节点都打了耗时日志并且把成功数、失败数、重试数通过计数器实时暴露到监控端点。这样即使没有专职监控平台也能通过定时抓取日志快速发现问题。数据同步不像在线接口用户不会立刻感受到它挂了所以被动等用户反馈和主动监控两侧都要有。另外命名规范和字段约定的力量一开始会被低估。dig1.0 里统一用 requestId 追踪一条数据从源到目标的完整路径所有时间字段都统一用标准格式所有任务名都用下划线连接。这个约定在排查问题时节省了大量沟通成本后续接入新同事时只需要看一份很短的规范文档就能上手。如果让我重做一遍我会在项目开始第一天就把“字段映射关系”单独定义成表而不是散落在代码里。当时为了赶进度很多字段转换逻辑直接写在 Transform 节点里导致后期调整一个字段名还要牵动代码。所以强烈建议数据源字段和目标字段的对应关系从一开始就配置化代码永远只负责执行不负责写死映射。dig1.0 不是什么庞大的系统但它确实让我意识到很多团队缺的不是数据处理框架而是一个能把数据可靠搬运的“管道工”。把基础工作做扎实比引入一套华丽的技术栈更能解决实际问题。
阅读完成 · 觉得有帮助？
