3、数据源连接优化：连接池配置、并行连接数、超时设置、重试策略

数据源连接这块，说白了就是你的摄取管道跟数据库、消息队列、API 这些外部系统之间的「握手」环节。我见过太多团队把精力全放在数据转换和写入上，结果连接层一塌糊涂——连接池炸了、超时没人管、重试把下游打崩。嗯，今天咱们就把这层彻底捋清楚。

3.1 连接池配置：别让「创建连接」吃掉你的时间

每次新建一个 TCP 连接，开销都不小。三次握手、TLS 协商、认证……一套下来几十毫秒就没了。如果每条数据都新建连接，吞吐直接腰斩。

我个人习惯用 HikariCP（Java 生态）或 psycopg2 的连接池（Python 生态）。核心参数就三个：

参数	说明	我的建议值
maximumPoolSize	池中最大连接数	CPU 核心数 × 2 + 1（经验公式）
minimumIdle	最小空闲连接数	通常等于 maximumPoolSize，避免冷启动
connectionTimeout	获取连接的最大等待时间	5000ms（5 秒）

代码示例（Java + HikariCP）：

HikariConfig config = new HikariConfig();
config.setJdbcUrl("jdbc:postgresql://host:5432/db");
config.setUsername("user");
config.setPassword("pass");
config.setMaximumPoolSize(20);
config.setMinimumIdle(20);
config.setConnectionTimeout(5000);
config.setIdleTimeout(300000);
config.setMaxLifetime(600000);

HikariDataSource ds = new HikariDataSource(config);

3.2 并行连接数：找到你的「甜蜜点」

并行连接数，就是同时从数据源拉数据的线程数。设少了，CPU 闲着；设多了，资源争抢、连接超时、数据库被打死。

我在项目中遇到过这样一个场景：从 Kafka 消费数据写入 ClickHouse，一开始只开了 4 个并行消费者，吞吐只有 2MB/s。后来逐步增加到 16 个，吞吐涨到 12MB/s。但再往上加到 32 个，吞吐反而掉到 8MB/s——因为 ClickHouse 的写入合并线程扛不住了。

怎么找到这个「甜蜜点」？我建议分三步走：

1. 压测基线：从 1 个连接开始，逐步增加，记录吞吐和延迟。
2. 观察拐点：当吞吐不再线性增长，甚至下降时，就是上限。
3. 留余量：把并行数设为拐点值的 80%。

3.3 超时设置：别让一个慢查询拖垮整个管道

超时设置，说白了就是给每个操作一个「死线」。超过这个时间，直接放弃，别死等。

我一般会设置三个层级的超时：

层级	参数	建议值	说明
连接超时	connectTimeout	5s	建立 TCP 连接的最大等待时间
读取超时	socketTimeout	30s	从连接读取数据的最大等待时间
查询超时	queryTimeout	60s	整个 SQL 查询的最大执行时间

代码示例（Python + psycopg2）：

import psycopg2
from psycopg2 import pool

connection_pool = psycopg2.pool.SimpleConnectionPool(
    1, 20,
    host="host",
    port=5432,
    dbname="db",
    user="user",
    password="pass",
    connect_timeout=5,
    options="-c statement_timeout=60000"
)

# 使用连接
conn = connection_pool.getconn()
conn.set_session(autocommit=True)
cur = conn.cursor()
cur.execute("SET socket_timeout = 30")
cur.execute("SELECT * FROM large_table LIMIT 1000")

3.4 重试策略：优雅地处理失败

重试不是简单的「失败了就再来一次」。不加控制的重试，就是 DDoS 攻击。

我曾经接手过一个数据管道，重试策略是「失败后立即重试，最多重试 10 次」。结果有一次数据库主库挂了，10 个并行线程同时重试，每秒产生 1000+ 个连接请求，直接把从库也打崩了。嗯，这就是典型的「好心办坏事」。

我推荐的重试策略是：指数退避 + 随机抖动。

• 指数退避：第一次重试等 1 秒，第二次等 2 秒，第三次等 4 秒……
• 随机抖动：在等待时间上加上随机偏移（比如 ±20%），防止所有客户端同时重试。

代码示例（Python + tenacity）：

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import psycopg2

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=30),
    retry=retry_if_exception_type((psycopg2.OperationalError, psycopg2.InterfaceError))
)
def fetch_data(query):
    conn = connection_pool.getconn()
    try:
        cur = conn.cursor()
        cur.execute(query)
        return cur.fetchall()
    finally:
        connection_pool.putconn(conn)

3.5 整体架构：四个维度如何协同

下面这张图展示了连接优化的四个维度如何协同工作。你想想看，连接池是「蓄水池」，并行连接数是「水龙头数量」，超时是「安全阀」，重试是「备用泵」——缺一不可。

3.6 实战配置清单

最后，我整理了一份实战配置清单。你直接拿去用，根据实际情况微调就行。

维度	参数	推荐值	备注
连接池	maximumPoolSize	CPU × 2 + 1	不超过数据库 max_connections 的 80%
连接池	minimumIdle	= maximumPoolSize	避免冷启动
连接池	connectionTimeout	5000ms	获取连接超时
并行连接数	parallelism	压测拐点的 80%	留余量，防止突发流量
超时	connectTimeout	5s	TCP 连接超时
超时	socketTimeout	30s	读取数据超时
超时	queryTimeout	60s	SQL 执行超时
重试	maxAttempts	3-5 次	太多会打爆下游
重试	backoff	指数退避 + 抖动	初始 1s，最大 30s