在Python的asyncio编程中，错误处理与同步代码存在显著差异。由于协程在事件循环中调度，一个未被妥善处理的异常可能不会立即抛出，而是附着在任务对象上，直到被显式获取才引发程序中断。理解这种异步错误传播路径，是构建稳定服务的前提。
协程内部的直接捕获
最基础的错误处理方式仍然是在协程函数中使用try except结构。当某个异步调用可能失败时，我们应当在其周围包裹异常捕获逻辑，避免异常向外冒泡至事件循环层造成无解崩溃。与同步代码不同的是，异步上下文中的资源清理最好配合async with或try finally使用，确保连接和文件句柄被正确释放。
下方示例展示了一个带有超时与异常捕获的异步请求函数。这里使用asyncio.wait_for来限制协程最长执行时间，并在捕获到超时或网络异常时返回统一的错误字典，而不是让异常逃逸。
import asyncio
async def fetch_data(url):
try:
async with asyncio.timeout(3):
# 模拟网络请求
await asyncio.sleep(1)
if url == 'bad':
raise ValueError('invalid url')
return {'status': 'ok', 'data': url}
except asyncio.TimeoutError:
return {'status': 'error', 'msg': 'timeout'}
except Exception as e:
return {'status': 'error', 'msg': str(e)}
这种写法的优势在于逻辑清晰、错误边界明确，每个协程对自己范围内的异常负责。其缺点是当协程数量极多时，重复的try except会造成一定模板代码冗余，且如果忘记捕获，异常仍会遗留到任务层。
任务级的异常监控
当协程通过asyncio.create_task被包装为Task后，异常会绑定到该Task对象。如果程序从未对Task调用result()或抛出异常，Python会在事件循环关闭时打印“Task exception was never retrieved”警告。为此，我们可以为任务添加add_done_callback，在回调中统一检查异常状态。
以下代码演示如何利用回调机制集中处理任务异常，而不必在每个协程内部都写冗余的返回结构。回调里通过task.exception()获取错误，若不为None则进行日志上报或补偿操作。
import asyncio
def handle_done(task):
exc = task.exception()
if exc is not None:
print(f'任务失败: {exc}')
else:
print(f'任务成功: {task.result()}')
async def main():
async def work(n):
if n == 2:
raise RuntimeError('任务2出错')
await asyncio.sleep(0.1)
return n
tasks = []
for i in range(4):
t = asyncio.create_task(work(i))
t.add_done_callback(handle_done)
tasks.append(t)
await asyncio.gather(*tasks)
asyncio.run(main())
该方案的优点是解耦了业务逻辑与错误处理，适合在框架层统一接入监控。但要注意，若同时使用gather等待这些任务，gather本身也会重新抛出首个异常，因此需配合return_exceptions或外层捕获来避免重复处理。
并发收集时的批量容错
在需要并发执行大量协程并希望部分失败不影响整体的场景中，asyncio.gather的return_exceptions参数非常关键。将其设为True后，异常会作为普通结果返回，而不是中断整个等待过程。调用方随后可遍历结果列表，区分成功值与异常实例。
下面示例并发查询多个接口，即使其中一个抛出错误，其余结果依然可用。最后通过类型判断把异常单独归类，实现高效的批量错误收敛。
import asyncio
async def query(i):
if i == 3:
raise ConnectionError('连接失败')
await asyncio.sleep(0.2)
return f'结果{i}'
async def main():
results = await asyncio.gather(
*(query(i) for i in range(5)),
return_exceptions=True
)
ok = [r for r in results if not isinstance(r, Exception)]
errs = [r for r in results if isinstance(r, Exception)]
print('成功:', ok)
print('失败:', errs)
asyncio.run(main())
使用return_exceptions能显著提升异步管道的鲁棒性，尤其适合爬虫、批量推送等允许局部失败的场合。其潜在问题是若疏忽了对返回值的类型检查，异常可能被当成正常数据流入后续逻辑，因此团队内部应约定明确的校验规范。
统一错误结构与日志
为了让异步系统中的错误可被观测，建议定义统一的错误封装，例如包含错误码、上下文与原始异常的结构体，并在上述各层捕获点转换为该结构。结合标准库logging，在任务回调或gather后统一输出，可避免异常信息散落。
实践中也可借助asyncio的Loop异常处理器，通过loop.set_exception_handler定制全局未捕获异常行为，将致命错误导向告警通道。这样从协程内、任务回调到事件循环三层防护，才能算得上高效的异步错误处理体系。
import asyncio
async def main():
loop = asyncio.get_running_loop()
def custom_handler(loop, context):
print('全局捕获:', context.get('exception'))
loop.set_exception_handler(custom_handler)
raise RuntimeError('测试全局异常')
asyncio.run(main())
综合来看，高效的异步错误处理并不是单点技巧，而是分层策略：协程内做基础防御，任务层做边界监控，并发层做批量容错，全局层做兜底。理清这些机制的协作方式，才能让Python异步程序既快又稳。