第36章：SQLAlchemy Async 内幕、绿丝与生产 SRE 手册

一、项目背景“线上异步 API 间歇性全部超时——错误全是QueuePool limit reached但连接数监控显示只有 30%——怎么回事”周四下午 3 点星云订单中台的异步服务突然全部 503。运维查 Prometheus 发现连接池checked_out从正常的 5 飙升到 25pool_size5, max_overflow20 最大 25——打满池的上限。重启服务后暂时恢复但 15 分钟后再次触发。排查发现根因是某个异步任务中调用了同步阻塞的time.sleep(5)。在 asyncio 的事件循环中time.sleep()会阻塞当前线程——导致事件循环停止调度其他协程。在这个 5 秒的阻塞窗口内已有的数据库连接没有被归还、新的连接请求排队——形成了连接池的拥塞坍塌。更糟糕的是pool_timeout超时的请求没有正确的重试机制直接被 503 返回给用户。另一个隐藏问题是greenlet_spawn的边界——create_async_engine的异步并不是全链路异步。它在内部通过 greenlet 把异步上下文桥接到同步的 DBAPIpsycopg路径——SQL 的实际执行仍然在同步线程中。如果 greenlet 所在的线程被阻塞如磁盘 I/O整个异步 Session 都会被卡住。本章将深入 asyncio SQLAlchemy 的协同机制greenlet 的工作原理、常见的五种生产故障模式池耗尽、长事务、隐式 IO、事件循环阻塞、迁移锁表并以此编写一份生产级《SQLAlchemy 故障 Runbook》。二、项目设计场景故障复盘会上大师在监控面板上标注了从第一个time.sleep阻塞到连接池满队列超时的完整时间线。小胖看着这条多米诺骨牌链条目瞪口呆。小胖“我就写了一个time.sleep(5)模拟网络延迟——怎么就把整个服务搞崩了”大师“因为在 asyncio 里time.sleep()是同步阻塞——它会冻结当前线程上的事件循环。在这 5 秒内事件循环不能调度任何其他协程——包括那些等待归还连接的协程。连接池里的连接被持有着新请求拿不到连接——排队。当队列长度超过pool_size max_overflow——超时。”小胖“那正确的写法是await asyncio.sleep(5)”大师“对。asyncio.sleep()会把控制权交还给事件循环——其他协程可以在这 5 秒内正常执行。这就是 asyncio 的核心规则永远不要在异步上下文中调用同步阻塞函数。”小白“技术映射time.sleep 马路中间停下来系鞋带后面全部堵死asyncio.sleep 开到路边停车区再系鞋带其他人正常通行。”大师“但 SQLAlchemy 的异步还有一个更隐蔽的问题——greenlet_spawn。”# create_async_engine 内部的工作原理# await session.execute(stmt)# → asyncio 事件循环# → greenlet_spawn(sync_fn) # 切换到 greenlet 线程# → sync_conn.execute(stmt) # 走同步 DBAPI 路径# → psycopg cursor.execute() # 阻塞式网络 I/O# → 结果返回 greenlet# → greenlet 结果传回 asyncio# → 事件循环继续小胖“技术映射greenlet 协作文书——它在 asyncio 和同步 DBAPI 之间做传话筒但自己也是在某个线程上跑的如果被阻塞了文书也罢工。”大师“这引出五个典型的生产故障模式——我今天要给你们一份完整的 Runbook。”故障模式二长事务小胖“长事务怎么会导致池耗尽事务一直在用同一个连接——不会归还池”大师“对。如果一个事务持续 30 秒比如批量处理 1000 个订单并逐条调外部支付网关连接一直被占用。其他请求只能排队——排队超时。解决方案是缩短事务持有时间——事务内只做数据库操作外部 IO 移到事务外。”故障模式三迁移锁表小白“Alembic 迁移时ALTER TABLE ... ADD COLUMN NOT NULL DEFAULT ...会锁表——这段时间内其他查询怎么办”大师“ALTER TABLE会获取AccessExclusiveLock——这是最重的锁。迁移期间该表上的所有 SELECT/INSERT/UPDATE 都会被阻塞。所以生产迁移必须分步先加 nullable 列 → 回填数据 → 再加约束——把锁表时间从几分钟降到几毫秒。”三、项目实战实战目标演示异步环境下的五种典型故障池耗尽、长事务、隐式 IO、事件循环阻塞、迁移锁表。为每种故障编写 Runbook 修复步骤与预防措施。步骤一故障模拟——异步池耗尽ch36_async_sre.py —— Async 内幕与生产 SRE 手册importasyncioimporttimefromsqlalchemy.ext.asyncioimport(create_async_engine,AsyncSession,async_sessionmaker,)fromsqlalchemyimporttext,select,String,Integerfromsqlalchemy.ormimportDeclarativeBase,Mapped,mapped_columnimportcontextvars# # 故障 1池耗尽——通过错误调用 time.sleep 复现# asyncdefreproduce_pool_exhaustion():复现异步任务中调用了同步阻塞 → 连接池耗尽enginecreate_async_engine(postgresqlasyncpg://nebula:nebula_devlocalhost:5432/order_center,pool_size2,# 很小的池——便于触发max_overflow3,# 最大 5 个连接pool_timeout5,# 排队 5 秒超时echoFalse,)Factoryasync_sessionmaker(engine,expire_on_commitFalse)asyncdefgood_task(task_id):正常任务正确使用 asyncio.sleepasyncwithFactory()ass:awaits.execute(text(SELECT 1))awaitasyncio.sleep(1)# ✅ 异步休眠——不阻塞事件循环awaits.execute(text(SELECT 2))returnfTask{task_id}: OKasyncdefbad_task(task_id):问题任务错误使用 time.sleepasyncwithFactory()ass:awaits.execute(text(SELECT 1))time.sleep(3)# ❌ 同步阻塞——冻住事件循环awaits.execute(text(SELECT 2))returnfTask{task_id}: OKprint( 故障 1异步池耗尽 )resultsawaitasyncio.gather(bad_task(1),# 持有连接 1 阻塞 3 秒bad_task(2),# 持有连接 2 阻塞 3 秒good_task(3),# 连接 3good_task(4),# 连接 4good_task(5),# 连接 5最后可用good_task(6),# 池满排队 → 超时或等待good_task(7),# 排队...return_exceptionsTrue,)forrinresults:ifisinstance(r,Exception):print(f ❌ 故障:{type(r).__name__}:{r})else:print(f ✅{r})awaitengine.dispose()# asyncio.run(reproduce_pool_exhaustion())步骤二故障模拟——长事务与隐式 IO# # 故障 2长事务——事务内调用外部 IO# LONG_TX_RUNBOOK 故障现象 连接池 checked_out 持续为 1不下降其他请求排队超时。 监控发现 pool_timeout 告警频繁。 根因分析 async with session.begin(): # 事务开始连接 checkout order await session.get(Order, 1) order.status processing await session.flush() # ↓↓↓ 以下在事务内调用了外部支付网关 result await payment_gateway.charge(order.total_amount) # 3-5 秒 if result.success: order.status paid await session.commit() # 连接在这个时刻才归还 else: await session.rollback() 事务持有连接的时间 数据库操作时间 外部 IO 时间 外部 IO 的 3-5 秒内连接被无偿占用——其他请求等不到连接。 修复步骤 1. 将外部 IO 移出事务边界 - 先 commit 订单状态为 processing - 再调用外部支付网关 - 支付成功 → 新事务中更新状态为 paid 2. 代码修复示例 # 事务 1设置初始状态 async with session.begin(): order.status processing await session.commit() # 事务外调用外部服务 pay_result await payment_gateway.charge(order.total_amount) # 事务 2更新最终状态 if pay_result.success: async with session.begin(): order await session.get(Order, 1) order.status paid await session.commit() 预防措施 - Code Review 中检查事务块内是否包含非数据库操作 - 在事件中记录事务开始/结束时间超过阈值如 500ms告警 - 使用 SQLAlchemy 的 after_commit 事件处理外部通知 print(\n 故障 2长事务 Runbook )print(LONG_TX_RUNBOOK)步骤三故障四——事件循环阻塞检测# # 故障 3事件循环阻塞检测# EVENT_LOOP_BLOCK_RUNBOOK 故障现象 服务响应 P99 偶尔飙升到 10 秒以上。 数据库连接池使用率正常慢 SQL 日志无异常。 症状间歇性出现——不可稳定复现。 根因分析 异步服务中某处在执行 CPU 密集型同步操作 如 JSON 大文件解析、图片处理、正则匹配百万行文本 阻塞了事件循环线程。其他协程被冻住——包括正常的 SQL 查询。 检测方法安装 asyncio 事件循环慢回调检测器 import asyncio loop asyncio.get_event_loop() loop.slow_callback_duration 0.1 # 超过 100ms 的回调会被记录 # Python 3.12 内置 # PYTHONASYNCIODEBUG1 python main.py 修复步骤 1. 定位阻塞代码用 loop.slow_callback_duration 或 py-spy 采样线程栈 2. 将 CPU 密集型操作移到线程池 result await loop.run_in_executor( None, # 默认线程池 cpu_intensive_function, arg1, arg2 ) 3. 或将 CPU 密集型任务移到独立的 Worker 进程Celery/ARQ 预防措施 - CI 中加入事件循环阻塞检测pytest-asyncio slow_callback_duration - 服务启动时自动启用 slow_callback_duration - 压测场景注入的模拟慢操作验证事件循环健康度 print( 故障 3事件循环阻塞 Runbook )print(EVENT_LOOP_BLOCK_RUNBOOK)步骤四故障五——迁移锁表与主从切换# # 故障 4迁移锁表# MIGRATION_LOCK_RUNBOOK 故障现象 执行 alembic upgrade 期间API 接口全部超时。 数据库 pg_stat_activity 显示大量 waiting for AccessExclusiveLock。 alembic 进程本身正常执行等待获取锁。 根因分析 ALTER TABLE ... ADD COLUMN ... NOT NULL DEFAULT ... → 获取 AccessExclusiveLock排他锁 → 表上的所有其他操作SELECT/INSERT/UPDATE/DELETE全部排队 → 有新事务在迁移开始前持有了该表的 ShareLock如一个正在运行的查询 → ALTER TABLE 等待该查询结束 ← 但该查询后又来了 100 个新查询排在 ALTER TABLE 后面 → 连锁阻塞新查询等 ALTER TABLEALTER TABLE 等旧查询 修复步骤 1. 迁移前检查锁状态 SELECT pid, state, wait_event_type, query FROM pg_stat_activity WHERE wait_event_type Lock AND query NOT LIKE %pg_stat%; 2. 设置 lock_timeout——超时后迁移主动放弃 op.execute(SET lock_timeout 5s) 3. 分步迁移——不一条 ALTER TABLE 加 NOT NULL ① op.add_column(orders, sa.Column(new_col, ...), nullableTrue) ② 数据回填分批 UPDATE ③ op.alter_column(orders, new_col, nullableFalse) 4. 生产变更窗口 - 提前通知业务方计划维护窗口低峰期如凌晨 2-4 点 - 迁移前在测试环境验证 upgrade/downgrade 循环 - 准备回滚预案alembic downgrade -1 备份验证 预防措施 - 禁止在高峰期执行 DDL - 在 alembic 迁移中强制加 lock_timeout - 迁移评审 Board 检查是否有 NOT NULL 大数据量 - 运维提前降低 API Worker 副本数或切流量 print( 故障 4迁移锁表 Runbook )print(MIGRATION_LOCK_RUNBOOK)# # 故障 5主从切换后连接池指向旧主机# FAILOVER_RUNBOOK 故障现象 数据库主从切换Failover完成后应用持续报 cannot execute UPDATE in a read-only transaction。 重启服务后恢复。 根因分析 主从切换后旧主库变为只读副本。连接池中已有的连接 仍指向旧主库的 IP——旧主库现在只能读。 新请求从池中拿到旧连接 → 执行 WRITE 操作 → 报 read-only error。 虽然 pool_pre_pingTrue 可以检测到连接断开 但 read-only transaction 错误不是连接断开——TCP 连接仍存活 pre_ping 发 SELECT 1 也能成功但写操作被拒。 修复步骤 1. 监控检测 read-only transaction 错误 - 在 handle_error 事件中捕获 ProgrammingError(read-only transaction) - 自动 dispose 当前引擎的连接池全部清空重建 - 或标记当前连接为非活跃下次请求时自动重建 2. 代码修复示例 event.listens_for(engine, handle_error) def handle_readonly_error(exception_context): if read-only in str(exception_context.original_exception).lower(): engine.pool.dispose() # 清空所有旧连接 # 新请求会创建新连接 → 指向新主库 3. 运维最佳实践 - 主从切换后触发应用的连接池重置如调用 /health/recycle 端点 - 使用 PgBouncer 前置代理——它对后端连接变化有更好的检测 预防措施 - 设置合理的 pool_recycle如 1800s 30分钟——连接定期回收 - pool_pre_pingTrue 配合乐观重试catch OperationalError retry - 运维在切换时主动调用各应用的健康检查接口触发连接池重建 print( 故障 5主从切换 Runbook )print(FAILOVER_RUNBOOK)步骤五完整的生产 SRE 手册# # 《SQLAlchemy 生产故障 Runbook》完整版# PRODUCTION_RUNBOOK ╔══════════════════════════════════════════════════════════════╗ ║ SQLAlchemy 生产故障 Runbook v2.0 ║ ╠══════════════════════════════════════════════════════════════╣ ║ 指标 │ 告警阈值 │ 排查命令 ║ ╠══════════════════════════════════════════════════════════════╣ ║ pool.checked_out │ 80% pool_size │ pool.checkedout() ║ ║ pool.overflow │ 0持续非零 │ pool.overflow() ║ ║ SQL 耗时 P99 │ 500ms │ after_cursor_execute ║ ║ SQL 错误率 │ 0.1% │ handle_error 事件 ║ ║ 事务持有时长 │ 1000ms │ before_commit 事件 ║ ║ 回滚率 │ 1% │ after_rollback 事件 ║ ║ 迁移锁等待 │ 5s │ pg_stat_activity ║ ╚══════════════════════════════════════════════════════════════╝ 通用修复步骤速查 [连接泄漏] 症状: checked_out 持续增长pool_size overflow 打满 优先级: P0立即处理 步骤: 1. 检查 pool.checkedout() 是否持续增长从不下降 2. 检查代码是否有 session Factory(); ... 无 close() 模式 3. 审查最近上线的变更是否有新增的 session 或 engine.connect() 未 close 4. 临时措施: 重启该 Worker 释放所有连接 5. 永久修复: 代码改用 with Factory() as s: 上下文管理器 [慢 SQL] 症状: P50 正常但 P99 飙升 优先级: P1 步骤: 1. 查看慢 SQL 日志中 EXPLAIN 输出 2. 检查是否 Seq Scan全表扫描→ 加索引 3. 检查是否 JOIN 过多 → 优化查询或加 covering index 4. 检查统计信息是否过期 → ANALYZE table [N1 雪崩] 症状: SQL 日志中同表独立 SELECT 激增 优先级: P1 步骤: 1. 设置 lazyraise_on_sql 暴露问题 2. 使用 selectinload / joinedload 预加载关联数据 3. 修复后重新启用 lazyselect默认 [连接池耗尽] 症状: QueuePool limit ... reached, connection timed out 优先级: P0 步骤: 1. 检查是否有长事务持续时间 1s 2. 检查是否有 time.sleep / 同步阻塞在异步路径中 3. 检查 max_overflow 是否足够 4. 检查数据库端连接数限制 (max_connections) [迁移锁表] 症状: API 超时pg_stat_activity 显示 Lock wait 优先级: P1 步骤: 1. 终止锁等待的查询SELECT pg_terminate_backend(pid) 2. 或终止正在执行的迁移如果尚未开始 DDL 3. 分步执行迁移先 nullable → 回填 → 再约束 4. 后续迁移设置 lock_timeout print(\n 生产 Runbook 完整版 )print(PRODUCTION_RUNBOOK)步骤六监控指标采集脚本# # 生产监控指标采集# fromprometheus_clientimportHistogram,Counter,Gauge,generate_latestfromfastapiimportFastAPI,Response# 指标定义sql_durationHistogram(sqlalchemy_sql_duration_seconds,SQL execution duration,[operation,table],buckets[0.01,0.05,0.1,0.25,0.5,1.0,2.5,5.0],)sql_errorsCounter(sqlalchemy_sql_errors_total,SQL execution errors,[error_type],)pool_checked_outGauge(sqlalchemy_pool_checked_out,Currently checked out connections,)pool_size_gaugeGauge(sqlalchemy_pool_size,Current pool size,)defregister_metrics(engine):为引擎注册 Prometheus 指标采集importtimefromsqlalchemyimporteventevent.listens_for(engine,before_cursor_execute)defstart_timer(conn,cursor,statement,params,ctx,em):conn.info[metric_start]time.monotonic()event.listens_for(engine,after_cursor_execute)defrecord_metric(conn,cursor,statement,params,ctx,em):startconn.info.pop(metric_start,None)ifstart:elapsedtime.monotonic()-start opstatement.strip().split()[0][:10]tableunknownstmt_lowerstatement.strip().lower()iffrominstmt_lower:tablestmt_lower.split(from)[-1].strip().split()[0][:30]sql_duration.labels(operationop,tabletable).observe(elapsed)event.listens_for(engine,handle_error)defrecord_error(exception_context):err_typetype(exception_context.original_exception).__name__ sql_errors.labels(error_typeerr_type).inc()# 定期更新池状态可在 /metrics 端点中调用defupdate_pool_metrics():poolengine.pool pool_checked_out.set(pool.checkedout())pool_size_gauge.set(pool.size())returnupdate_pool_metricsprint(\n 监控指标采集 )print( 已定义指标: sql_duration, sql_errors, pool_checked_out, pool_size)print( 暴露端点: GET /metrics → Prometheus 拉取)可能遇到的坑及解决方法greenlet_spawn的递归深度限制现象深度嵌套的异步查询链A→B→C→D在 greenlet 层面可能触发RecursionError。解决减少嵌套深度或使用session.run_sync()显式切换到同步路径。异步环境下pool.checkedout()在 sync_engine 上不可用现象async_engine.pool.checkedout()返回的是AsyncAdaptedQueuePool的属性。解决使用async_engine.sync_engine.pool.checkedout()获取真实值。pool_pre_ping在 asyncio 中的额外开销被放大现象每次 checkout 多一次的SELECT 1在网络延迟 10ms 的环境下占总耗时的 50%。解决如果网络延迟极低同 AZ 1ms开销可忽略。跨 AZ 考虑pool_recycle。测试验证# tests/test_ch36_async_sre.pyimportpytestimportasynciofromsqlalchemy.ext.asyncioimportcreate_async_engine,AsyncSession,async_sessionmakerfromsqlalchemyimporttext,eventpytest.mark.asyncioasyncdeftest_pool_exhaustion_detected():验证池耗尽可以被检测enginecreate_async_engine(sqliteaiosqlite:///:memory:,pool_size1,max_overflow0,echoFalse)Factoryasync_sessionmaker(engine,expire_on_commitFalse)asyncwithFactory()ass:# 正常操作——可以拿到连接awaits.execute(text(SELECT 1))# 模拟池耗尽检查poolengine.sync_engine.poolassertpool.checkedout()0,连接应已归还awaitengine.dispose()pytest.mark.asyncioasyncdeftest_long_transaction_detection():验证可以通过事件检测长事务enginecreate_async_engine(sqliteaiosqlite:///:memory:,pool_size5,echoFalse)long_tx_detected[]event.listens_for(engine.sync_engine,commit)defdetect_long_tx(conn):passevent.listens_for(engine.sync_engine,before_cursor_execute)deftrack_start(conn,cursor,statement,parameters,context,executemany):importtime conn.info[tx_start]time.monotonic()Factoryasync_sessionmaker(engine,expire_on_commitFalse)asyncwithFactory()ass:awaits.execute(text(SELECT 1))awaits.commit()assertTrue# 不会超时awaitengine.dispose()四、项目总结五大故障模式速查故障症状根因修复优先级池耗尽QueuePool limit reached连接泄漏/长事务/阻塞P0长事务checked_out 不下降事务内调外部 IOP1隐式 IO偶发超时time.sleep / 文件 IO 阻塞循环P1事件循环阻塞P99 偶发飙升CPU 密集同步操作P2迁移锁表所有 API 超时ALTER TABLE 排他锁P0适用场景异步 Web 服务FastAPI SQLAlchemy asyncio——Runbook 作为运维操作手册。联合压测——在低峰期模拟五种故障模式验证恢复流程。新人培训——用 Runbook 案例讲解 asyncio 与同步阻塞的微妙区别。混沌工程——用 Chaos Mesh / Gremlin 注入连接池故障测试系统韧性。不适用场景纯同步 Python 服务create_engine而非create_async_engine——池耗尽和阻塞问题性质不同。注意事项pool.checkedout()在异步引擎上要通过sync_engine访问。Runbook 中的修复方案需要先在测试环境验证——不要在生产环境直接尝试。asyncio.create_task()创建的后台任务可能被意外取消——确保正确使用try/finally归还连接。常见踩坑经验案例 1使用了concurrent.futures.ThreadPoolExecutor但线程内的 session 未正确关闭现象线程池线程中的 session 在任务结束后未 close——连接永不归还。修复在每个线程任务的finally块中调用session.close()。案例 2async for遍历大型结果集时持有连接的时间过长现象async for row in await session.stream(stmt)在处理每行时耗时 10ms1000 行走完 10 秒——连接被持有 10 秒。修复使用yield_per(N)fetchmany()分段消费或把处理逻辑移到消费循环外。案例 3多个asyncio.gather共享一个 Session 导致并发冲突现象两个协程共享同一个 AsyncSession 对象一个在查询、一个在 flush——报ConcurrentModificationError。修复每个协程使用独立的 Sessionasync_sessionmaker()()。思考题greenlet_spawn本质上是在 asyncio 的事件循环线程中运行一个 greenlet它在该线程上执行同步 DBAPI 调用。如果同步 DBAPI 调用被网络 IO 阻塞如cursor.execute()等待数据库响应greenlet 会让出 CPU 吗如果不会它与真正的全链路异步如直接用 asyncpg 的 async API在吞吐量上有什么差异在微服务架构中一个服务可能有多个数据库主库 只读副本 租户隔离库。连接池的参数应该为每个数据库独立设置还是使用统一参数如果独立设置如何在 Runbook 中表示不同数据库的告警阈值延伸阅读与资源NumPy 从入门到生产落地全链路实战指南科学计算/向量化Redis 8 实战精讲从 CRUD 到源码构建高可用缓存系统Redis 实战修炼与原理进阶Python 3实战精进从脚本到高并发订单引擎python入门Rquests从菜鸟脚本到企业级SDK的网络实战圣经Milvus向量数据库实战修炼从 0 到 1精通向量检索与生产落地MongoDB 实战进阶与内核修炼后端工程师的 AI 转型第一课Ollama 与私有化大模型实战10倍开发者的 Dify 魔法书从零构建全栈 AI 应用后端工程师转型AI第一课-Ollama 与私有化大模型实战大型语言模型(LLM) vLLM 高性能推理落地实战Agent开发之LlamaIndex 实战修炼与源码进阶大语言模型Transformers 实战修炼与源码剖析