📚 课程目录第4章数据源连接与适配

4、数据源连接与适配：关系型数据库、API、消息队列（Kafka）、文件系统的连接配置

数据流水线跑不起来，十有八九是源头出了问题。我这些年排查过的故障里，数据源连接失败占了将近一半。不是密码过期，就是网络策略变了，再不然就是驱动版本不匹配。说白了，连接配置这块看似简单，坑却不少。

这一章咱们就聊聊四种最常见的数据源——关系型数据库、API、Kafka 和文件系统——怎么配、怎么调、怎么避坑。

4.1 关系型数据库：JDBC 连接的那些事

关系型数据库是流水线里最传统的数据源。MySQL、PostgreSQL、Oracle，基本都走 JDBC 协议。配置上大同小异，但有几个关键点我建议你盯紧。

4.1.1 连接字符串与驱动

先看一个典型的 MySQL 连接配置：

# application.properties
spring.datasource.url=jdbc:mysql://10.0.1.100:3306/data_pipeline?useSSL=false&serverTimezone=Asia/Shanghai&rewriteBatchedStatements=true
spring.datasource.username=pipeline_user
spring.datasource.password=EncryptedPassword123!
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
spring.datasource.hikari.maximum-pool-size=20
spring.datasource.hikari.minimum-idle=5
spring.datasource.hikari.connection-timeout=30000
spring.datasource.hikari.idle-timeout=600000
spring.datasource.hikari.max-lifetime=1800000

嗯，这里要注意几个参数：

useSSL=false：内网环境一般关掉，省去证书验证的麻烦。但生产环境如果走公网，必须开。
rewriteBatchedStatements=true：批量写入时性能提升明显，我实测过，能快 3-5 倍。
connection-timeout：设 30 秒比较合理。太短了网络抖动就断，太长了故障恢复慢。

注意：密码不要硬编码在配置文件里。我曾经接手过一个项目，密码直接写在 Git 仓库里，结果被扫描工具扫出来，差点出事。建议用环境变量或密钥管理服务（如 Vault、AWS Secrets Manager）。

4.1.2 连接池调优

连接池不是越大越好。我见过有人把 maximum-pool-size 设成 200，结果数据库直接被压垮。为什么？因为每个连接都要占用数据库端的内存和线程资源。

我的经验公式是这样的：

场景	推荐连接数	说明
轻量查询（OLTP）	10-20	每次查询毫秒级返回
批量写入（ETL）	5-10	长连接，占用时间长
混合负载	15-30	根据实际监控调整

小技巧：连接池的 max-lifetime 要设得比数据库端的 wait_timeout 短。比如 MySQL 默认 wait_timeout 是 8 小时，那 max-lifetime 设 30 分钟就够了。这样连接被数据库回收前，连接池自己先关掉，避免连接泄漏。

4.2 API 数据源：RESTful 与 GraphQL

API 数据源现在越来越常见。特别是微服务架构下，很多数据都要通过 HTTP 接口拉取。配置 API 连接时，我主要关注三件事：认证、限流、重试。

4.2.1 认证方式

常见的认证方式有这几种：

：最简单，放在 Header 里。但安全性一般，适合内部服务。
Basic Auth：用户名密码 Base64 编码，配合 HTTPS 使用。
OAuth 2.0：最常用，需要先获取，然后带访问。有过期时间，要处理刷新逻辑。
JWT：自包含的，服务端不需要存 session，适合分布式场景。

我建议优先用 OAuth 2.0 客户端凭证模式（Client Credentials）。配置示例：

# api-source.yml
api:
  base-url: https://api.example.com/v2
  auth:
    type: oauth2
    grant-type: client_credentials
    client-id: ${API_CLIENT_ID}
    client-secret: ${API_CLIENT_SECRET}
    -url: https://auth.example.com/oauth/
  retry:
    max-attempts: 3
    backoff: exponential
    initial-interval: 1000
  rate-limit:
    requests-per-second: 10
    burst: 20

4.2.2 限流与重试策略

调用外部 API 最怕被限流。我曾经遇到过，一个批处理任务因为没做限流，直接把对方服务打挂了，结果被对方运维找上门来……

我的做法是：

客户端限流：用令牌桶算法，控制每秒请求数。比如设 10 QPS，突发允许 20。
指数退避重试：第一次失败等 1 秒，第二次等 2 秒，第三次等 4 秒。最多重试 3 次。
熔断机制：连续失败超过阈值（比如 5 次），直接熔断 30 秒，不再发送请求。

核心原则：API 调用一定要做幂等处理。同一个请求发两次，结果应该一样。这样重试时才不会产生脏数据。

4.3 消息队列（Kafka）：消费者的配置艺术

Kafka 作为数据源，配置比数据库复杂得多。我见过太多人因为配置不当，导致数据重复消费、消费延迟、甚至 OOM。

4.3.1 消费者核心参数

先看一个生产级的消费者配置：

# kafka-consumer.properties
bootstrap.servers=kafka1:9092,kafka2:9092,kafka3:9092
group.id=data-pipeline-group
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
enable.auto.commit=false
auto.offset.reset=earliest
max.poll.records=500
max.poll.interval.ms=300000
session.timeout.ms=45000
heartbeat.interval.ms=15000
fetch.min.bytes=1024
fetch.max.wait.ms=500

几个关键点：

enable.auto.commit=false：手动提交偏移量。自动提交容易丢数据，我建议你关掉。
auto.offset.reset=earliest：新消费者组从最早的消息开始消费。如果改成 latest，新加入的消费者会跳过历史数据。
max.poll.records=500：每次拉取最多 500 条。设太大容易 OOM，设太小吞吐量上不去。

4.3.2 偏移量管理

偏移量管理是 Kafka 消费的核心。我的做法是：

消费完一批数据后，先处理业务逻辑。
处理成功，再手动提交偏移量。
处理失败，不提交，下次重新消费。

代码示例：

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> record : records) {
        try {
            processRecord(record);
        } catch (Exception e) {
            log.error("处理消息失败，将重试: {}", record.key(), e);
            // 记录到死信队列或重试队列
            sendToDlq(record);
        }
    }
    // 手动提交
    consumer.commitSync();
}

注意：commitSync 是同步提交，会阻塞。如果对延迟敏感，可以用 commitAsync。但 commitAsync 有回调失败的风险，建议两者结合：异步提交 + 定期同步提交兜底。

4.4 文件系统：本地、HDFS 与对象存储

文件系统数据源看起来简单，但坑也不少。特别是文件编码、路径格式、权限问题，我踩过不少。

4.4.1 常见文件系统对比

类型	协议/接口	适用场景	注意事项
本地文件系统	file://	单机测试、小规模	路径分隔符、权限
HDFS	hdfs://	大数据批处理	NameNode 高可用、块大小
S3 / OSS	s3a:// / oss://	云原生、弹性扩展	访问密钥、区域、限速
SFTP / FTP	sftp://	外部系统对接	连接池、被动模式

4.4.2 文件监听与增量读取

文件系统做数据源，核心是增量读取。我常用的策略：

按时间戳：记录上次读取的时间，只读取修改时间晚于该时间的文件。
按文件名模式：比如 data_20240101.csv，按日期滚动。
按目录结构：/data/year=2024/month=01/day=01/，按分区读取。

配置示例（基于 Apache Camel 的文件监听）：

from("file:///data/incoming?include=*.csv&move=.done&delay=5000")
    .process(exchange -> {
        // 处理文件
        InputStream is = exchange.getIn().getBody(InputStream.class);
        processCsv(is);
    });

经验之谈：文件处理完一定要做标记。要么移动到 .done 目录，要么重命名加 .processed 后缀。否则下次扫描又会读到，造成重复处理。我曾经因为这个原因，把同一批数据灌了三次，查了好久才发现。

4.5 知识体系总览

下面这张图总结了四种数据源的核心配置要点和常见故障点，你可以对照着检查自己的流水线：

数据源连接配置，说白了就是三个字：稳、快、准。连接要稳定，不能动不动就断；速度要快，不能成为流水线的瓶颈；数据要准，不能丢也不能重复。你配置的时候，多想想这三个维度，基本不会出大问题。