Psycopg3连接池遇‘connection is closed’错误：重连与预防方案咨询

你的问题根源在于连接池中的连接可能因数据库端超时（比如PostgreSQL的idle_in_transaction_session_timeout、tcp_keepalives_idle设置）或网络波动变得无效，但连接池仍会返回这些失效连接。以下是针对性的解决方法：

1. 获取连接时主动校验有效性 #

从连接池获取连接后，先执行简单SQL验证连接可用性，若失败则重新获取。修改装饰器实现：

def provide_db(func):
    """
    Function decorator that provides a session if it isn't provided.
    """

    @wraps(func)
    def wrapper(*args, **kwargs):
        arg_session = 'db'

        func_params = func.__code__.co_varnames
        session_in_args = arg_session in func_params and \
                          func_params.index(arg_session) < len(args)
        session_in_kwargs = arg_session in kwargs

        if session_in_kwargs and kwargs[arg_session] is not None:
            return func(*args, **kwargs)
        if session_in_args and args[func_params.index(arg_session)] is not None:
            return func(*args, **kwargs)
        else:
            max_retries = 2
            for _ in range(max_retries):
                with master_connection_pool.connection() as conn:
                    try:
                        # 执行简单查询校验连接状态
                        conn.execute("SELECT 1")
                        conn.row_factory = dict_row
                        kwargs[arg_session] = conn
                        return func(*args, **kwargs)
                    except psycopg.OperationalError:
                        # 连接失效，跳过当前连接重试
                        continue
            raise RuntimeError("无法获取有效的数据库连接")

    return wrapper

2. 配置连接池自动健康检查 #

Psycopg3的ConnectionPool支持通过参数设置连接校验逻辑，初始化时添加相关配置：

def check_connection(conn):
    try:
        conn.execute("SELECT 1")
        return True
    except psycopg.OperationalError:
        return False

master_connection_pool = psycopg_pool.ConnectionPool(
    conninfo=masterDbConnectionString,
    min_size=5,
    max_size=10,  # 建议设置上限避免连接过载
    open=True,
    check=check_connection,  # 获取连接时自动校验有效性
    reset=lambda conn: conn.rollback(),  # 归还连接前回滚未提交事务
)

连接池会自动过滤失效连接，确保返回的连接可用。

3. 优化数据库端超时配置 #

如果是数据库主动关闭闲置连接，可调整PostgreSQL参数：

• 调大idle_in_transaction_session_timeout（默认10分钟），避免事务中闲置过久被关闭
• 启用tcp_keepalives_idle、tcp_keepalives_interval参数，保持TCP连接活跃，防止网络设备断开连接

额外提示：修复SQL注入风险 #

你的示例代码用f-string拼接SQL存在严重安全漏洞，应改用参数化查询：

@provide_db
def do_Stuff(user_id, db):
    db.execute("SELECT * FROM users WHERE id = %s", (user_id,))
    ...

内容的提问来源于stack exchange，提问作者Frank Conry