asyncio.create_task() 无法直接监控 asyncpg 慢查询，因 fetch/execute 不暴露耗时且不支持钩子；需通过封装 Connection 或装饰器+contextvars 实现无侵入计时与告警。
为什么 asyncio.create_task() 不能直接套 asyncpg.fetch() 就算完
异步数据库慢查询告警不是“加个超时”就能生效的——asyncpg 的 fetch()、execute() 等方法本身不暴露执行耗时，也不支持在调用时传入全局钩子。你直接用 asyncio.create_task() 包一层，只会让任务并发跑起来，但完全无法感知它卡在哪、卡了多久。
真正可行的路径是：拦截 SQL 执行入口，在执行前打点、执行后计算耗时、超阈值就触发告警。这需要利用 asyncpg 提供的 Connection.add_listener() 或更底层的 Connection._protocol？错，它不适用。正确入口是 Connection.execute() 这类方法的封装层。
- 必须自己封装一层
asyncpg.Connection子类或代理对象，重写fetch()、execute()等关键方法 - 所有数据库操作必须走这个封装层，不能直接 import asyncpg 后裸调用
- 告警逻辑（如发 Slack、写日志）要异步非阻塞，否则会拖垮整个 event loop
- 注意连接池场景：每个
Pool.acquire()返回的Connection都得被包装，不是只包 pool 对象
如何用装饰器 + contextvars 实现无侵入的慢查询计时
比起继承 Connection，更轻量的做法是用装饰器包裹所有执行方法，并借助 contextvars.ContextVar 记录当前请求上下文（比如 trace_id），避免多任务间耗时统计串扰。
核心难点在于：asyncpg 的方法是协程函数，装饰器必须是 async def 形式，且需在 await 前后精确计时：
立即学习“Python免费学习笔记（深入）”；
import time
import asyncio
from contextvars import ContextVar
<p>_slow_threshold_var = ContextVar('slow_threshold_ms', default=500.0)</p><div class="aritcle_card flexRow">
<div class="artcardd flexRow">
<a class="aritcle_card_img" href="/xiazai/skill8910" title="361SCAN security check your skill python scripts"><img
src="https://img.php.cn/upload/skill/000/000/081/179231049778675.jpg" alt="361SCAN security check your skill python scripts" onerror="this.onerror='';this.src='/static/lhimages/moren/morentu.png'" ></a>
<div class="aritcle_card_info flexColumn">
<a href="/xiazai/skill8910" title="361SCAN security check your skill python scripts">361SCAN security check your skill python scripts</a>
<p>扫描并分析已安装技能，用于查看指定技能目录的名称、描述和详情，或列出所有已安装的技能。</p>
</div>
<a href="/xiazai/skill8910" title="361SCAN security check your skill python scripts" class="aritcle_card_btn flexRow flexcenter"><b></b><span>下载</span> </a>
</div>
</div><p>def with_slow_alert(func):
async def wrapper(*args, *<em>kwargs):
start = time.perf_counter()
try:
result = await func(</em>args, *<em>kwargs)
return result
finally:
elapsed_ms = (time.perf_counter() - start) </em> 1000
if elapsed_ms > _slow_threshold_var.get():</p><h1>这里发告警，注意别 await 阻塞！</h1><pre class="brush:php;toolbar:false;"> asyncio.create_task(_log_slow_query(func.__name__, elapsed_ms))
return wrapper
-
time.perf_counter()比time.time()更准，适合测量短时异步操作 - 不要在
finally里直接await _log_slow_query()，否则会阻塞当前协程；用asyncio.create_task()脱离当前上下文 -
ContextVar可让你按请求动态调整阈值，比如对后台批处理设为 5000ms，对 API 接口设为 300ms - 该装饰器必须作用于实际调用的协程方法上，比如包装
MyConnection.fetch()，而不是包装asyncpg.Connection.fetch()原生方法（它不可 patch）
asyncpg 连接池中怎么确保每个 connection 都被监控
很多人以为给 asyncpg.create_pool() 加个参数就能全局生效，其实不行。Pool 返回的是未包装的原始 Connection 对象，每次 acquire() 都是新实例。
正确做法是在获取连接后、使用前立即包装：
async with pool.acquire() as conn:
monitored_conn = MonitoredConnection(conn) # 自定义包装类
await monitored_conn.fetch("SELECT * FROM users WHERE id = $1", 123)
- 不要试图 monkey patch
asyncpg.Connection类——它的方法是 Cython 实现，patch 失败率高且不可靠 - 推荐用组合而非继承：
MonitoredConnection内部持有一个conn: asyncpg.Connection，所有方法都转发并加监控 - 如果用 ORM（如 TortoiseORM 或 GINO），优先查它是否支持自定义 executor 或 query hook；GINO 有
before_cursor_execute事件，比手动包装 connection 更干净 - 注意 connection 关闭时，包装对象也要透传
close()和terminate()，否则连接泄漏
告警发送失败会导致 event loop 卡死吗
会，如果你在慢查询回调里同步发 HTTP 请求（比如用 requests.post()），或者 await 一个没设 timeout 的 aiohttp 请求，整个 event loop 就可能被 hang 住，后续所有异步任务停摆。
- 告警必须用带 timeout 的异步 client，例如
aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=2)) - 建议加 fallback：告警失败时仅写本地文件或内存 ring buffer，避免雪崩
- 高频慢查询（比如每秒几十次）要限流，用
asyncio.Semaphore(3)控制并发告警数，防止打爆告警通道 - 别忘了记录 SQL 本身——但注意脱敏：
str(query).replace(user_id, '***')，否则日志泄露敏感信息
最易被忽略的一点：慢查询阈值不该写死在代码里。上线后不同环境（dev/staging/prod）负载差异大，必须从配置中心或环境变量加载，且支持热更新——否则某次 DB 升级后索引失效，你得重启服务才能调低阈值。
