3. 数据源接入与适配：打通企业数据的第一公里

数据共享平台建好了，最头疼的问题来了——数据怎么进来？

我见过太多项目，平台架构设计得漂漂亮亮，结果卡在数据接入这一步。说白了，企业里的数据源五花八门，有跑了好几年的 MySQL，有为了高并发上的 Redis，还有业务部门自己搭的 Kafka 管道。每个数据源都有自己的脾气。

这一章，我们就来聊聊怎么把这些「脾气各异」的数据源，稳稳当当地接入共享平台。

3.1 关系型数据库接入：MySQL 与 Oracle

关系型数据库是企业数据的「老大哥」。大部分核心业务数据，都躺在这些库里。我个人习惯，先把它们分成两类：OLTP 和 OLAP。MySQL 和 Oracle 通常承担 OLTP 角色，接入时最怕的就是把源库拖垮。

3.1.1 全量同步 vs 增量同步

全量同步很简单，就是一次性把数据全拉过来。但要注意，千万不能在业务高峰期跑全量。我曾经有个同事，下午三点直接跑了个全量同步，结果源库 CPU 飙到 99%，业务直接挂了半小时……嗯，后来他再也不敢了。

增量同步才是日常主力。常用的方式有这几种：

方式	原理	适用场景	坑点
时间戳	根据更新时间字段筛选	有统一更新时间字段的表	时间不准、时区问题
自增ID	记录最大ID，拉取新数据	只增不删的表	物理删除后ID断层
Binlog监听	解析数据库变更日志	实时性要求高的场景	Binlog格式、磁盘占用
触发器	数据库内建触发机制	小规模、可控环境	性能损耗大，不推荐

我个人最推荐 Binlog 监听。拿 MySQL 来说，Canal 这个组件就干这个事。它伪装成 MySQL 的从库，读取 Binlog，然后把变更事件推给下游。Oracle 的话，可以用 Oracle GoldenGate 或者 LogMiner。

3.1.2 连接池与参数调优

接入多个数据源时，连接池管理是个大活。我建议统一用 HikariCP，性能确实好。配置上注意几点：

• 最大连接数： 别贪多。一般 10-20 个就够，太多反而把源库拖死。
• 连接超时： 设成 30 秒。超过就报错，别死等。
• 空闲检测： 每 5 分钟检测一次，把无效连接清掉。

# 一个典型的 HikariCP 配置示例
spring.datasource.hikari.maximum-pool-size=15
spring.datasource.hikari.connection-timeout=30000
spring.datasource.hikari.idle-timeout=600000
spring.datasource.hikari.max-lifetime=1800000
spring.datasource.hikari.connection-test-query=SELECT 1

3.2 NoSQL 数据源接入：MongoDB 与 Redis

NoSQL 数据库接入，思路和关系型不太一样。它们的数据模型更灵活，但接入方式也更「野」。

3.2.1 MongoDB：文档型数据的接入

MongoDB 存的是 BSON 文档，结构多变。接入时，我建议用 Change Streams 做实时同步。它比轮询高效得多，能直接监听集合的变更事件。

但有个坑——Change Streams 依赖副本集。如果你用的是单节点 MongoDB，那就没法用了。这时候只能退而求其次，用定时查询 + oplog 的方式。

// MongoDB Change Streams 接入示例
MongoClient mongoClient = MongoClients.create("mongodb://localhost:27017");
MongoDatabase database = mongoClient.getDatabase("myDB");
MongoCollection<Document> collection = database.getCollection("orders");

ChangeStreamIterable<Document> changeStream = collection.watch();
changeStream.forEach(event -> {
    Document fullDocument = event.getFullDocument();
    // 将变更数据推送到共享平台
    pushToDataPlatform(fullDocument);
});

3.2.2 Redis：缓存数据的接入

Redis 接入，说实话，有点特殊。它主要是缓存，数据量通常不大，但变更极快。我一般不把 Redis 当「主数据源」接入，而是作为 实时数据管道 的一环。

常用的方式：

• Keyspace Notification： 监听 key 的过期、删除事件。适合缓存失效后的补偿同步。
• Redis Stream： Redis 5.0 引入的消息队列能力。可以作为轻量级数据管道。
• 定时全量拉取： 如果 Redis 存的是配置类数据，每小时拉一次就够了。

我记得有个项目，业务方非要把 Redis 里的用户 Session 实时同步到数据平台。我劝了半天——Session 数据生命周期短，同步过去也没分析价值。后来他们听了，改成只同步登录日志，效果反而更好。

3.3 消息队列接入：Kafka

Kafka 接入，是数据共享平台的「标配」。几乎所有实时数据，最终都会流经 Kafka。

3.3.1 Consumer 组与分区分配

接入 Kafka，核心是 Consumer 的设计。我见过太多人，上来就写一个 Consumer，然后抱怨 Kafka 慢。其实问题出在分区分配上。

记住一个公式：Consumer 数量 ≤ 分区数量。如果 Consumer 比分区多，多出来的就是闲置的，白白浪费资源。

// Kafka Consumer 接入示例
Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092,kafka2:9092");
props.put("group.id", "data-platform-group");
props.put("enable.auto.commit", "false");  // 手动提交偏移量
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("order-topic", "user-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        processRecord(record);
    }
    consumer.commitSync();  // 处理完再提交
}

3.3.2 序列化与反序列化

Kafka 里的数据，建议统一用 Avro 或 Protobuf。JSON 虽然方便，但体积大、解析慢。我个人的经验，Avro + Schema Registry 是最稳的组合。Schema 变更时，Registry 能保证上下游兼容。

3.4 API 数据源接入

API 接入，是最灵活也最「脏」的方式。外部系统、SaaS 服务、老系统，往往只给你一个 HTTP 接口。

3.4.1 接入策略

API 接入，我总结了三个要点：

• 限流： 调用方有限制，自己也要有限制。用令牌桶算法，每秒最多 N 次。
• 重试： 网络抖动是常态。指数退避重试，最多重试 3 次。
• 幂等： 同一个请求，发多少次结果都一样。防止重复数据。

# 一个简单的 API 接入重试逻辑（伪代码）
def fetch_api_data(url, retries=3):
    for i in range(retries):
        try:
            response = requests.get(url, timeout=10)
            if response.status_code == 200:
                return response.json()
        except (ConnectionError, Timeout) as e:
            if i == retries - 1:
                raise e
            time.sleep(2 ** i)  # 指数退避
    return None

3.4.2 数据格式转换

API 返回的数据，格式千奇百怪。有的嵌套 5 层，有的字段名是中文。接入后，我建议统一做一次 ETL 清洗：字段映射、类型转换、空值处理。别把脏数据直接丢进共享平台。

3.5 整体接入架构

说了这么多，我们来看看整体架构长什么样。下面这张图，是我在多个项目中沉淀下来的接入层设计。

你看，整个架构分四层：数据源层、适配器层、统一接入层、数据转换层。每一层各司其职，出了问题也能快速定位。我个人习惯，每个适配器都独立部署成微服务，这样某个数据源挂了，不会影响其他数据源的接入。