数据库专题23连接池与超时——让数据库故障停在边界内“数据库挂了”并不等于接口必须一直卡住。很多事故是连接没有超时、线程被耗尽最后连健康检查都无法响应。本篇用 PostgreSQL、Redis、MongoDB 三种客户端做一次完整的连接池设计启动时创建、请求间复用、关闭时释放为连接、获取连接和单条命令分别设置上限最后用故障注入验证服务能否在有限时间内失败。上一篇练习讲解审计不能相信请求体上一篇要求为评论删除和文章冻结定义事件并证明普通作者不能伪造管理员角色。正确做法是从 PostgreSQL 的current_user或会话对象取得actor_id与role请求体只允许提交reason。审计写入重复时用event_id幂等不能用upsert覆盖旧事件。这个原则在连接池也适用连接参数由服务器配置决定不能让客户端请求任意修改超时时间。1. 三类连接为什么都需要池建立 TCP/TLS、认证和数据库会话都有成本。每次请求临时创建连接会在高并发时同时触发大量握手完全不限制连接数又会超过 PostgreSQL 的max_connections。连接池在两者之间做排队池满时请求等待但等待必须有 deadline。请求 - acquire(最多100ms) - 执行(最多500ms) - release │超时 └─ 返回 503并带 request_id不继续占用 worker池大小不是越大越好。假设应用有 4 个 worker、PostgreSQL 允许 80 个连接单实例池通常设置 10~15并为迁移、监控和后台任务预留余量。Redis/Mongo 也要限制并发尤其是事务或大文档查询。2. PostgreSQL SQLAlchemy 连接池项目使用 SQLAlchemy 2.x。pool_size是常驻连接数max_overflow是短时额外连接pool_timeout是等待连接的最长时间connect_args的connect_timeout则控制 TCP 建连。fromsqlalchemyimportcreate_engine,text enginecreate_engine(postgresqlpsycopg://blog:bloglocalhost:5432/blog,pool_size10,max_overflow5,pool_timeout2,# 池排队超过2秒立即失败pool_recycle1800,# 防止云数据库空闲连接被中途回收connect_args{connect_timeout:2},)defcheck_db()-int:只执行最轻量的探活查询探活失败不能触发无限重连。withengine.connect()asconn:returnconn.execute(text(select 1)).scalar_one()事务必须用上下文管理器异常时自动回滚并把连接归还池defpublish_article(article_id:int)-None:withengine.begin()asconn:# 成功 commit异常 rollbackconn.execute(text(update articles set statuspublished where id:id),{id:article_id})不要在全局保存一个正在使用的Connection那会让多个请求共享事务状态出现“别人的更新被我提交”的灾难。3. Redis 客户端的异步池importredis.asyncioasredis redis_poolredis.ConnectionPool.from_url(redis://localhost:6379/0,max_connections50,socket_connect_timeout1,socket_timeout0.5,health_check_interval30,)cacheredis.Redis(connection_poolredis_pool,decode_responsesTrue)asyncdefread_cached(key:str)-str|None:缓存读取失败时抛出给 service 层由 service 决定是否回源。returnawaitcache.get(key)asyncdefclose_cache()-None:awaitcache.aclose()Redis 命令超时要短于 HTTP 请求超时例如接口总预算 800msRedis 只给 150ms剩余时间留给 PostgreSQL 回源和序列化。不要在异常捕获里立刻循环重试三次否则 150ms 会变成 450ms并发故障更严重。4. MongoDB 的 Server Selection 与池frompymongoimportMongoClient mongoMongoClient(mongodb://localhost:27017/?replicaSetrs0,maxPoolSize30,minPoolSize3,waitQueueTimeoutMS200,connectTimeoutMS1000,serverSelectionTimeoutMS1000,socketTimeoutMS1500,)deflatest_version(article_id:int)-dict|None:serverSelectionTimeoutMS 确保主节点不可用时一秒内返回错误。returnmongo.blog.article_versions.find_one({article_id:article_id},sort[(version,-1)])waitQueueTimeoutMS控制池已满时等待多久serverSelectionTimeoutMS控制寻找可用节点的时间两者不是同一个概念。副本集场景还应设置readPreference历史读取可以从 secondary 读取但恢复操作必须使用 primary。5. 把异常转换成用户能理解的响应importasynciofromfastapiimportHTTPExceptionasyncdefsafe_cache_get(key:str):try:returnawaitasyncio.wait_for(cache.get(key),timeout0.2)except(asyncio.TimeoutError,redis.RedisError):# 日志记录 request_id返回 None 让上层回源而不是返回假数据returnNonedefdb_503(exc:Exception)-HTTPException:returnHTTPException(status_code503,detail{code:database_unavailable,message:服务暂时繁忙请稍后重试},)所有请求入口都要有总超时例如 3 秒但事务内部不应任意取消。异步取消发生在数据库命令刚提交之后可能让客户端以为失败而实际已成功所以写操作要使用幂等键并在重试时查询结果。6. 故障注入实验先让客户端连接池只有一个连接再启动两个并发任务第一个故意pg_sleep(1)第二个应在pool_timeout到达时失败。importconcurrent.futuresfromsqlalchemyimporttextdefslow_query()-None:withengine.connect()asconn:conn.execute(text(select pg_sleep(1)))withconcurrent.futures.ThreadPoolExecutor(max_workers2)aspool:futures[pool.submit(slow_query)for_inrange(2)]forfinfutures:try:f.result()exceptExceptionasexc:print(type(exc).__name__,exc)验收时记录第二个请求是否在约 2 秒内结束、进程是否仍能执行select 1、日志是否带 request_id。停止 Mongo 或 Redis 后重复实验不能出现无限重试和线程数持续增长。常见排查Too many connections统计所有实例的pool_size max_overflow不要只看单个容器。连接归还后仍被占用检查是否漏写with以及流式查询是否关闭结果集。Redis 读超时但接口 30 秒才返回检查是否在异常处理里叠加重试。Mongo 查询偶发慢确认是否命中索引并区分waitQueueTimeoutMS与socketTimeoutMS。课后练习把database-platform的三个客户端封装成resources.py提供startup()与shutdown()写一个测试把 PostgreSQL 池大小设为 1证明第二个任务会超时而不是无限等待为 Redis 回源路径记录耗时和降级次数。下一篇进入数据库账号、最小权限和注入防护。