本文目录导读：
Python脚本中协程请求失败的优雅处理：从重试机制到异常隔离
目录导读
- 协程请求失败的常见场景与本质
- 基于asyncio的异常捕获与重试策略
- 断路器模式：避免无效重试的系统雪崩
- 请求失败时的降级与回退方案
- 【问答】协程失败处理高频问题与最佳实践
- 构建鲁棒的异步请求处理链路
协程请求失败的常见场景与本质
在异步编程中,协程的请求失败往往比同步代码更加隐蔽。核心挑战在于： 协程的并发特性使得单个请求的失败可能影响整个事件循环的稳定性，常见失败场景包括：
- 网络超时（timeout）
- 服务端返回5xx错误
- 连接池耗尽（Connection pool exhausted）
- SSL证书验证失败
- 异步库底层异常（如aiohttp的ClientError）
本质原因：Python的async/await语法虽然简化了异步代码的编写，但异常传播路径与传统线程模型不同，一个未捕获的协程异常会沿着await链向上冒泡，最终可能导致asyncio.run()崩溃或任务泄漏。
关键认知：协程失败处理不是简单的try-except，而是需要结合任务调度策略与资源隔离。
基于asyncio的异常捕获与重试策略
1 基础异常捕获模式
import asyncio
import aiohttp
async def fetch_url(session, url, retries=3):
for attempt in range(retries):
try:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response:
response.raise_for_status() # 自动引发HTTP异常
return await response.json()
except aiohttp.ClientError as e:
if attempt == retries - 1:
raise # 最后一次失败，向上传播
await asyncio.sleep(2 ** attempt) # 指数退避
2 高级重试配置
使用第三方库tenacity可以极大简化重试逻辑：
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=0.5, max=10),
retry=retry_if_exception_type((aiohttp.ServerTimeoutError, aiohttp.ClientResponseError))
)
async def robust_fetch(session, url):
async with session.get(url) as response:
response.raise_for_status()
return await response.read()
注意：当重试所有次数仍失败时，tenacity会抛出RetryError，该异常包含最后一次尝试的原始异常信息。
断路器模式：避免无效重试的系统雪崩
当目标服务持续不可用时,无限制的重试会加剧系统负担。断路器模式在协程中实现如下：
class CircuitBreaker:
def __init__(self, failure_threshold=5, recovery_timeout=30):
self.failure_count = 0
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self._state = 'CLOSED' # CLOSED, OPEN, HALF_OPEN
self._last_failure_time = None
async def call(self, coroutine_func, *args, **kwargs):
if self._state == 'OPEN':
if time.time() - self._last_failure_time > self.recovery_timeout:
self._state = 'HALF_OPEN'
else:
raise CircuitBreakerOpenError("Service temporarily unavailable")
try:
result = await coroutine_func(*args, **kwargs)
if self._state == 'HALF_OPEN':
self._state = 'CLOSED'
self.failure_count = 0
return result
except Exception:
self.failure_count += 1
self._last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self._state = 'OPEN'
raise
实战价值：当断路器打开后，后续请求在阈值时间内直接返回错误，避免无效网络I/O，保护调用方的资源。
请求失败时的降级与回退方案
1 静默降级
对于非关键请求（如日志上报、埋点），失败降级为“记录日志后跳过”：
async def safe_report_analytics(session, data):
try:
await session.post('https://example.analytics.com/event', json=data)
except Exception as e:
logger.warning(f"Analytics failed, skip: {e}")
2 缓存回退
当主请求失败时,尝试从过期缓存或备用数据源获取数据：
async def get_user_profile(session, user_id):
try:
data = await fetch_from_api(session, user_id)
await cache.set(user_id, data, expire=3600)
return data
except Exception:
cached = await cache.get(user_id)
if cached:
return cached
raise # 无缓存时仍视为失败
3 并发请求的隔离
使用asyncio.gather时，设置return_exceptions=True避免单个失败影响全部：
async def fetch_multiple(urls):
async with aiohttp.ClientSession() as session:
tasks = [fetch_single(session, url) for url in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)
# 筛选异常
successes = [r for r in results if not isinstance(r, Exception)]
failures = [r for r in results if isinstance(r, Exception)]
return successes, failures
【问答】协程失败处理高频问题与最佳实践
Q1: 为什么try-except在协程中有时抓不到异常？
A: 协程中的异常遵循延迟计算的特性，如果使用asyncio.create_task启动任务但没有await它，异常会存储在任务对象中直到被await或显式task.exception()调用。最佳实践：对于fire-and-forget类型的任务，使用task.add_done_callback记录异常。
Q2: asyncio请求超时应该在哪一层设置？
A: 设置在三层：
- 传输层：操作系统socket超时（
socket.setdefaulttimeout） - 库级超时：
aiohttp.ClientTimeout - 协程级超时：
asyncio.wait_for(your_coro(), timeout=5)
推荐同时设置库级和协程级超时，前者控制单个请求耗时，后者兜底整个协程执行。
Q3: 重试时如何避免重复写入副作用？
A: 使用幂等性设计：
- 在请求中添加全局唯一ID（
idempotency-key） - 服务端校验重复请求
- 或者重试时改用GET请求（若原请求为POST）
Q4: 并发量高时如何避免重试风暴？
A: 实施限流+断路器组合策略：
- 使用
asyncio.Semaphore控制并发上限 - 动态调整重试间隔（指数退避+随机抖动）
- 实现令牌桶或漏桶算法控制整体请求速率
构建鲁棒的异步请求处理链路
协程请求失败处理的核心不是捕获每一个异常,而是根据业务重要性等级设计差异化策略：
| 请求类型 | 处理策略 |
|---|---|
| 关键数据查询 | 指数退避重试 + 断路器 + 缓存回退 |
| 非关键上报 | 异步日志记录后静默丢弃 |
| 批量并发 | 使用gather分离成功与失败，统一监控 |
最后的技术要点：
- 区分网络级异常与应用级异常，对前者重试，对后者上报错误日志
- 异步代码中的
finally块尤为重要——确保资源释放（如数据库连接、锁解锁） - 使用结构化日志（如structlog）记录每次失败的原因与重试次数，便于事后复盘
通过将这些策略内化到你的Python脚本中,协程请求失败将从“混乱的源头”转变为“可控的风险”，让你的异步系统真正具备生产级稳定性。
