1. 为什么需要关注协程异常处理在异步编程的世界里协程就像是一群在厨房里忙碌的厨师。当你在Python中使用asyncio时每个协程都像是一位正在烹饪的厨师而事件循环则是厨房主管负责协调谁该在什么时候使用炉灶。但问题来了——如果某位厨师把锅烧糊了抛出异常或者一道菜做的时间太长超时整个厨房就会陷入混乱。我曾在实际项目中遇到过这样的场景一个基于asyncio的爬虫程序因为某个协程中的网络请求超时未处理导致整个程序挂起数小时而不自知。更糟的是由于异常未被正确捕获程序在失败后甚至无法正常记录日志。这就是为什么我们需要系统性地掌握协程的异常处理和超时控制机制。2. 协程异常处理的基础机制2.1 协程中的异常传播特性与普通函数不同协程中的异常不会立即向上冒泡。当一个协程内部抛出异常时这个异常首先会被挂起直到该协程被await时才会真正抛出。看看这个典型例子async def faulty_coro(): raise ValueError(Something went wrong) async def main(): try: coro faulty_coro() # 这里不会抛出异常 await coro # 异常在这里抛出 except ValueError as e: print(fCaught: {e})这个特性经常让初学者困惑。我在实际代码审查中发现很多开发者会错误地认为创建协程对象时就会执行代码而实际上它只是返回一个可等待对象。2.2 基础异常捕获模式对于协程的异常处理最基本的模式就是在await语句周围包裹try-except块。但有几个关键细节需要注意协程内部也可以有自己的try-except块形成多层次的异常处理使用asyncio.create_task()创建的任务需要单独处理异常未被处理的协程异常会导致运行时警告这里有个更完整的示例async def nested_failure(): try: await asyncio.sleep(1) raise RuntimeError(Inner error) except asyncio.CancelledError: print(Cancelled!) raise async def main(): try: await nested_failure() except RuntimeError as e: print(fOuter caught: {e})重要提示永远不要简单地捕获所有异常(Exception)。在异步环境中这可能会意外吞掉重要的asyncio.CancelledError导致任务无法正常取消。3. 任务级别的异常处理3.1 任务与异常回调当你使用asyncio.create_task()创建任务时异常处理方式发生了变化。任务中的异常不会自动传播除非你显式地await这个任务。这是很多bug的来源。async def failing_task(): await asyncio.sleep(1) raise ValueError(Task failed) async def main(): task asyncio.create_task(failing_task()) await asyncio.sleep(2) # 异常被存储在任务中但不会抛出 # 正确的处理方式 task asyncio.create_task(failing_task()) try: await task except ValueError as e: print(fTask failed: {e})更健壮的做法是添加回调函数def task_done_callback(task): if task.exception(): print(fTask failed with: {task.exception()}) async def main(): task asyncio.create_task(failing_task()) task.add_done_callback(task_done_callback) await asyncio.sleep(2)3.2 任务组模式Python 3.11引入了asyncio.TaskGroup它提供了更优雅的任务异常处理方式async def worker(name): await asyncio.sleep(1) if name bad: raise ValueError(f{name} failed) print(f{name} succeeded) async def main(): try: async with asyncio.TaskGroup() as tg: tg.create_task(worker(good)) tg.create_task(worker(bad)) except* ValueError as eg: # 注意这个新语法 for exc in eg.exceptions: print(fCaught: {exc})这种模式会自动取消所有任务当其中一个失败时是管理相关任务组的理想选择。4. 超时控制机制详解4.1 基本超时模式asyncio提供了几种实现超时控制的方式。最简单的是asyncio.wait_for()async def slow_operation(): await asyncio.sleep(10) return Done async def main(): try: result await asyncio.wait_for(slow_operation(), timeout1.0) except asyncio.TimeoutError: print(Operation timed out)但这种方法有个潜在问题即使超时发生了后台任务可能仍在运行。在我的一个Web爬虫项目中这导致了大量僵尸请求堆积最终耗尽了连接池。4.2 更健壮的超时模式更安全的做法是结合asyncio.shield()和任务取消async def critical_section(): try: await asyncio.sleep(10) print(Critical work done) except asyncio.CancelledError: await asyncio.sleep(2) # 清理工作 print(Cleanup complete) raise async def main(): task asyncio.create_task(critical_section()) try: await asyncio.wait_for(asyncio.shield(task), timeout1.0) except asyncio.TimeoutError: print(Timeout occurred, but cleanup will continue) # 此时原始任务仍在运行4.3 超时工具函数实现在实际项目中我通常会实现一个更灵活的超时工具函数async def with_timeout(coro, timeout, *, cancelTrue): task asyncio.create_task(coro) try: return await asyncio.wait_for(task, timeouttimeout) except asyncio.TimeoutError: if cancel: task.cancel() try: await task # 等待取消完成 except asyncio.CancelledError: pass raise这个实现允许选择是否在超时后取消任务并且确保任务被正确清理。5. 常见陷阱与最佳实践5.1 异常处理中的反模式在代码审查中我经常看到这些错误做法忽略任务异常# 不好的做法 asyncio.create_task(risky_operation())过度宽泛的异常捕获try: await coro() except Exception: # 会吞掉CancelledError pass不正确的超时嵌套# 可能导致混乱的超时行为 await asyncio.wait_for(asyncio.wait_for(coro(), timeout1), timeout2)5.2 调试技巧当协程出现异常时调试可能比较困难。我常用的几种方法启用asyncio调试模式asyncio.run(main(), debugTrue)检查未完成的任务pending asyncio.all_tasks() for task in pending: if task.done() and task.exception(): print(fAbandoned task failed: {task.exception()})使用结构化日志记录async def logged_coro(): try: await operation() except Exception as e: logger.error(Operation failed, exc_infoe) raise5.3 性能考量异常处理和超时控制会带来一定的性能开销。在需要高性能的场景下我有几个优化建议避免在热路径上使用频繁的超时检查对于大量并行任务考虑使用asyncio.gather()的return_exceptions参数超时时间不要设置得过短通常不少于100ms6. 实际项目中的应用示例6.1 Web请求重试机制在我的一个API客户端项目中实现了这样的重试逻辑async def fetch_with_retries(session, url, retries3, timeout5.0): for attempt in range(retries): try: async with session.get(url, timeouttimeout) as response: return await response.json() except (aiohttp.ClientError, asyncio.TimeoutError) as e: if attempt retries - 1: raise await asyncio.sleep(2 ** attempt) # 指数退避6.2 数据库事务处理处理数据库事务时尤其需要注意异常async def save_data(pool, data): async with pool.acquire() as conn: async with conn.transaction(): try: await conn.execute(INSERT..., data) await external_service_call(data) # 可能失败 except Exception: # 事务会自动回滚 logger.exception(Failed to save data) raise6.3 批量任务处理处理批量任务时的模式async def process_batch(tasks, batch_size10, timeout30.0): results [] for i in range(0, len(tasks), batch_size): batch tasks[i:ibatch_size] try: batch_results await asyncio.wait_for( asyncio.gather(*batch, return_exceptionsTrue), timeouttimeout ) results.extend(batch_results) except asyncio.TimeoutError: logger.warning(fBatch {i//batch_size} timed out) raise return results在长期运行的生产环境中良好的异常处理和超时控制是系统稳定性的基石。我建议为所有关键协程添加适当的监控和日志记录这样当问题发生时你至少能知道发生了什么而不是面对一个无声无息的挂起程序。