连接池把一次物理建连的成本分摊到多次使用中，但不会让建连成本消失，也不保证创建池时就已经打开连接。先分清两个对象：业务代码短期持有的借用句柄，以及底层真正连接数据库的连接。关闭前者，通常只是把后者归还池中，等待下一次复用，而不是立即断开数据库连接。

复用是一段生命周期，不是免费执行的承诺

远程数据库建连可能包含 TCP 握手、TLS、认证和会话初始化。复用能够减少每个请求重复执行这些步骤的次数，但 SQL 本身仍然需要时间。连接过期、断线、显式失效或进程替换之后，建连成本还会重新出现。一个进程内应长期复用其拥有的 SQLAlchemy Engine；如果每次请求都重新创建 Engine，那么每次也在重新创建自己的池，原本可以复用的资源就被拆散了。

SQLAlchemy 的池按需增长。配置 pool_size=2 的 QueuePool 起初是空的，并不意味着已经预热两条连接。先后执行的两次借用可以一直使用同一条物理连接；同时发生的借用才可能促使池增长。本例允许保留两条连接，再临时溢出一条，所以有限的 max_overflow=1 对应最多三条同时借用。全部归还后，最多留下两条空闲连接，而不是一直维持三条。

这些参数语义属于具体实现，不应套用到所有连接池。QueuePool 的 pool_size=0 表示取消大小限制，而不是关闭池化；max_overflow=-1 则允许无限溢出。需要限制数据库连接预算时，必须核对所用库的准确契约，并写出明确、有限的值，不能看到零就自行理解为禁用。

坏掉的样子

池被占满，可能是实际并发超过容量，也可能是长事务或借用后未归还。业务看到的总延迟因此需要拆成获取连接的等待与执行查询的时间。pool_timeout 限制的是等待可用槽位，不是新建物理连接的完整耗时，也不是 SQL 执行时间。实验用了很短的浮点数超时，便于触发失败，但这不是精确到几十毫秒的硬实时承诺，不能据此推导整个请求的截止时间。

归还也不等于抹掉上一个借用者留下的所有状态。SQLAlchemy 的默认归还重置通过回滚清理事务状态，并与 Connection.close() 协同完成；已经提交的临时表等会话状态仍可能保留。不要仅凭“用了上下文管理器，连接已经还回去”就推断租户隔离成立。应用需要哪些会话初始化与清理，应按具体数据库的能力和自身契约明确规定。

断线则是另一个边界。显式 invalidate 会丢弃当前底层 DBAPI 连接，之后使用逻辑 Connection 时，再从池取得替代连接。这个替代品可能是另一条现成空闲连接，不一定刚刚新建。如果失效时已有事务，必须先结束该事务才能重连；重连也不是自动恢复一个可能已经产生效果的事务，不能把连接生命周期管理当成业务重试保证。

快速验证

在 macOS 或 Linux 的空目录执行 python3 -m venv .venv，再执行 .venv/bin/python -m pip install SQLAlchemy==2.0.52。把代码保存为 pool_probe.py，用 .venv/bin/python pool_probe.py 运行。本次验证环境是 Python 3.14.4 与 SQLite 3.53.0。脚本自行创建并清理临时数据库，不连接任何数据库服务器，也不使用生产凭据。

from pathlib import Path
import sqlite3
from tempfile import TemporaryDirectory

import sqlalchemy
from sqlalchemy import create_engine, event
from sqlalchemy.exc import TimeoutError as PoolTimeout
from sqlalchemy.pool import QueuePool


assert sqlalchemy.__version__ == "2.0.52"
with TemporaryDirectory(prefix="pool-reuse-") as directory:
    engine = create_engine(
        "sqlite:///" + str(Path(directory) / "probe.db"),
        poolclass=QueuePool, pool_size=2, max_overflow=1, pool_timeout=0.05,
    )
    events = {"opened": 0, "closed": 0}

    @event.listens_for(engine, "connect")
    def opened(dbapi_connection, entry):
        events["opened"] += 1
        entry.info["serial"] = events["opened"]

    @event.listens_for(engine, "close")
    def closed(dbapi_connection, entry):
        events["closed"] += 1

    try:
        assert engine.pool.checkedin() == 0 and events["opened"] == 0
        with engine.begin() as connection:
            serial = connection.info["serial"]
            connection.exec_driver_sql("CREATE TEMP TABLE marker(value INTEGER)")
            connection.exec_driver_sql("INSERT INTO marker VALUES (42)")
        with engine.connect() as connection:
            assert connection.info["serial"] == serial
            assert connection.exec_driver_sql("SELECT value FROM marker").scalar_one() == 42
        assert events["opened"] == 1 and engine.pool.checkedout() == 0
        print("lazy start: 0 physical connections; two sequential leases reuse 1; committed temp state persists")

        held = []
        try:
            for _ in range(3):
                held.append(engine.connect())
            assert len({connection.info["serial"] for connection in held}) == 3
            assert engine.pool.checkedout() == 3 and events["opened"] == 3
            try:
                unexpected = engine.connect()
            except PoolTimeout:
                pass
            else:
                unexpected.close()
                raise AssertionError("the fourth concurrent lease must time out")
            assert events["opened"] == 3
        finally:
            for connection in held:
                connection.close()
        assert engine.pool.checkedout() == 0 and engine.pool.checkedin() == 2
        assert events["closed"] == 1
        print("2 persistent slots + 1 overflow allow 3 leases; the fourth times out; only 2 remain idle")

        with engine.connect() as connection, engine.connect() as occupied:
            previous = connection.info["serial"]
            occupied_serial = occupied.info["serial"]
            connection.invalidate()  # No transaction has started on this lease.
            assert connection.exec_driver_sql("SELECT 1").scalar_one() == 1
            assert connection.info["serial"] != previous
            assert occupied.info["serial"] == occupied_serial
            assert connection.exec_driver_sql(
                "SELECT count(*) FROM sqlite_temp_master WHERE name = 'marker'"
            ).scalar_one() == 0
        assert events["opened"] == 4
        print("with the other slot occupied, invalidation forces a new physical connection without the old temp table")
    finally:
        engine.dispose()
    assert events == {"opened": 4, "closed": 4}
print("all 4 connect events have matching close events; SQLite", sqlite3.sqlite_version)

第一行输出通过 connect 事件和保存在底层连接上的序号，证明池从零开始、两次顺序借用只创建一条物理连接。第一次借用提交的临时标记在下一次仍然可见，因此事务回滚式重置并不是完整会话重置。接着同时保留三份借用，第四次获取抛出 PoolTimeout；归还之后只剩两条空闲连接，并出现一次 close 事件。这同时区分了借用数量、可保留的空闲数量和真实建连次数。

失效检查特意先占住另一条空闲连接，保证后续 SQL 不能直接借到现成的替代品，因而确实触发新建物理连接。新连接的临时模式中没有旧标记。如果不占住那个槽位，池完全可以借出另一条空闲连接，因此“invalidate 必然建连”的断言就是错的。最后先归还全部借用，再 dispose 池，观察到四次 connect 与四次 close 事件对应。这里统计的是库事件，不是对任意后端关闭失败情形的资源回收强保证。

这个本地实验验证生命周期和容量，并没有测量 TCP、TLS 的节省时间，也不是吞吐基准。真正调整生产池时，应同时记录建连频率、获取等待、持有时长和查询延迟，辨别时间到底花在哪里。增大池必须仍然满足共享数据库的预算；它不能单独修好泄漏，也不能让慢查询自动变快。

来源