使用encode/databases库的数据库连接问题：应用闲置数小时后触发pymysql.err.InternalError错误

这个pymysql.err.InternalError: Packet sequence number wrong - got 0 expected 1错误很典型，本质是MySQL服务端主动断开了闲置超时的连接，但你的连接池没有检测到连接已失效，仍然复用了这个无效连接。结合你的技术栈，给你几个实用的解决思路：

1. 配置连接池的自动回收机制 #

databases模块的连接池其实支持自动回收过期连接，只是默认没开启合适的参数。你需要在初始化Database时指定pool_recycle参数，设置一个比MySQL服务器wait_timeout短的值，让连接池主动回收旧连接，避免被服务端断开。

修改你的db_config.py：

db = databases.Database(
    HOST,
    pool_recycle=300,  # 5分钟后自动回收连接，根据你的MySQL超时设置调整
    pool_size=5,       # 常驻连接数，按需调整
    max_overflow=10    # 峰值时允许的额外临时连接数
)

• pool_recycle：连接在池中存活的最大秒数，到期后会被自动关闭并重建，建议设置为MySQLwait_timeout的70%左右（比如MySQL默认8小时，你可以设3600秒=1小时）。
• pool_size：保持打开的常驻连接数量，根据你的并发需求调整。
• max_overflow：超出常驻连接数后，允许临时创建的连接数，用完后会被销毁。

2. 捕获连接错误并自动重试 #

即使配置了连接回收，偶尔还是可能出现连接在回收前被服务端断开的情况。你可以在事务装饰器里捕获特定的连接错误，自动重试一次操作。

修改你的perform_transaction装饰器：

from pymysql.err import InternalError

def perform_transaction(foo):
    @wraps(foo)
    async def decorate(*args, **kwargs):
        retry_limit = 1  # 重试1次，可按需调整
        for attempt in range(retry_limit + 1):
            try:
                async with db.connection() as connection:
                    async with connection.transaction():
                        res = await foo(*args, **kwargs)
                        return res
            except InternalError as e:
                error_msg = str(e)
                if "Packet sequence number wrong" in error_msg:
                    if attempt < retry_limit:
                        logger.warning("检测到失效连接，正在重试操作...")
                        continue
                    logger.error("连接重试失败: %s", error_msg)
                logger.error("数据库操作错误: %s", error_msg)
            except Exception as e:
                logger.error("未知错误: %s", str(e))
        return []
    return decorate

这样当检测到连接序列错误时，会自动重试一次，复用连接池的新连接完成操作。

3. 验证MySQL服务器的超时设置 #

先确认你的MySQL服务器的闲置超时参数，确保pool_recycle的值小于这些参数：
执行以下SQL命令查看：

SHOW VARIABLES LIKE 'wait_timeout';
SHOW VARIABLES LIKE 'interactive_timeout';

默认情况下这两个参数都是8小时（28800秒），你可以根据业务需求调整，或者把pool_recycle设为3600秒（1小时），这样更安全。

4. 可选：手动校验连接有效性 #

如果上面的方法还不够，你可以在获取连接后先执行一个简单的查询来验证连接是否有效：

async with db.connection() as connection:
    # 执行简单查询校验连接
    await connection.execute("SELECT 1")
    async with connection.transaction():
        res = await foo(*args, **kwargs)
        return res

这个方法会增加一点额外的查询开销，但能确保每次使用的都是有效连接，适合对稳定性要求极高的场景。

内容的提问来源于stack exchange，提问作者Oleksandr Valentirov