Python 超时处理完全指南

直接回答：任何与外部系统交互的 Python 代码都必须设超时——HTTP 请求、数据库连接、异步任务无一例外。超时是防止级联故障的安全阀：依赖方卡死时，应用可以停止等待并执行清理或降级，但是否终止底层工作取决于驱动、取消机制及远端服务。

为什么超时是刚需

没有超时的程序等于把命运交给了别人：下游 API 挂了，你的 worker 线程一个个挂起在等待上，连接池耗尽、内存上涨、整个服务雪崩——一个第三方的故障就这样变成了你的故障。超时的本质是：给"等待"设一个上限，超时后还需显式处理失败、取消与资源清理。线程里的阻塞函数未必随等待超时而停止，远端写操作也可能已完成。

HTTP 请求超时

import httpx

# 精细化：连接、读、写、池分开设
timeout = httpx.Timeout(10.0, connect=5.0)
resp = httpx.get("https://api.example.com/data", timeout=timeout)

# requests 也支持（单值或 (连接, 读取) 元组）
import requests
requests.get("https://api.example.com/data", timeout=(3.05, 27))

连接超时管"连不上"（网络/DNS 问题，应短），读取超时管"连上了不回话"（按接口正常耗时放宽）。这些通常是连接或相邻网络读写的阶段性等待限制，并非整个请求严格的总时长。端到端截止时间还需上层预算管理。

数据库超时

# PostgreSQL + psycopg 3；dsn、url 来自受控配置
import psycopg
from sqlalchemy import create_engine

with psycopg.connect(dsn, connect_timeout=5) as conn:
    with conn.cursor() as cur:
        cur.execute("SET LOCAL statement_timeout = '10s'")
        cur.execute("SELECT 1")
# 异常离开连接上下文时回滚；超时错误不能在失败事务中直接继续执行

# SQLAlchemy + PostgreSQL psycopg 方言与默认 QueuePool
engine = create_engine(url, connect_args={"connect_timeout": 5},
                       pool_timeout=10)

三层都设：连接超时（连库）、语句超时（慢查询自杀）、池超时（等不到连接就报错而非无限排队）。

异步超时

import asyncio

async def with_timeout():
    try:
        async with asyncio.timeout(5):     # Python 3.11+ 推荐写法
            return await slow_call()
    except TimeoutError:
        return "降级结果"

# 老版本：await asyncio.wait_for(slow_call(), timeout=5)

asyncio.timeout 上下文管理器清晰表达了"这块代码最多跑多久"；嵌套时先到期的截止时间先触发取消，并不总是内层优先。取消是协作式的，阻塞事件循环或吞掉 CancelledError 都会影响效果。AnyIO 的 move_on_after/fail_after 使用同步 with（即使在 async 函数中），不是 async with。

超时值怎么定

拍脑袋是事故源头，正确姿势是按数据定：

1. 测量正常值：依赖接口的 p95/p99 延迟是多少；
2. 留出余量：结合调用方延迟目标、依赖分布与可接受误超时率留余量，不机械套用固定倍数；
3. 预算分解：调用链上各层超时应该内小外大（下游 3s < 网关 5s < 前端 10s），否则外层先超时内层白等；
4. 特殊场景例外：批处理、报表这类本来就慢的操作单独放宽，别用全局值硬套。

超时之后做什么

超时只是"快速失败"，配套动作决定系统韧性：

• 重试：幂等操作 + 指数退避（1s、2s、4s），非幂等操作谨慎；
• 降级：缓存旧数据、默认值兜底，让核心功能活下去；
• 熔断：连续超时后暂停调用一段时间，给下游喘息也给自己省资源；
• 告警：超时率是依赖健康的晴雨表，必须可观测。

记录超时时，带上调用目标、耗时、重试次数和请求标识，避免只留一句 timeout。若这些日志经 DataKit进入观测云，可按依赖和版本核对超时集中在哪一段时间；结合客户端超时配置与下游日志判断是否值得重试，不能仅靠超时条数判定下游故障。

常见问题（FAQ）

Q：设了超时还是被拖垮？
A：检查是否每层都设了（连接/读/池/全局），以及重试是否放大流量（每次 5s、首次加重试 3 次就是 4 次尝试，还要计入退避和清理时间）。超时与重试要联合设计。

Q：0 超时或无限超时有合理场景吗？
A：先看库的语义，0 不一定代表无限，也可能表示立即超时；不少 API 用 None 表示禁用超时。流式长连接（SSE/WebSocket）可用心跳、空闲超时与业务截止时间共同管理；本地 Unix socket 调用可适当放宽，但仍建议有上限。

Q：超时引发的半截写操作怎么办？
A：这就是幂等设计的意义——超时后重试同一操作不会产生重复副作用。写接口设计时把幂等键（idempotency key）安排上。

官方参考

本文基于官方文档整理，未进行运行时或性能测试。示例中的业务函数、数据模型和部署地址需结合项目补全；局部片段不等同于完整生产应用。

• 官方资料 1
• 官方资料 2
• 官方资料 3
• 官方资料 4

延伸阅读

• Python subprocess 子进程完全指南
• Python 类型注解完全指南