Python asyncio 并发 I/O…

用 asyncio 的协程、任务、同步原语和异步上下文管理器，把大量 I/O 密集型操作安全地并发化，并加入限流、取消、重试、断路与优雅关闭。

适用场景

解决思路

1. 采用单线程协作式多任务模型编写程序：一个事件循环调度多个任务，任务只在 await 处主动让出控制权。
2. 使用非阻塞 I/O 和 I/O 多路复用同时处理多个连接，避免在等待某个 I/O 时阻塞其他任务。
3. 使用 asyncio.run(my_coroutine()) 运行顶层协程；Python 3.7+ 优先用这种方式。
4. 需要手动管理事件循环时，使用 asyncio.get_event_loop() 或 asyncio.new_event_loop() 获取/创建循环，asyncio.set_event_loop(loop) 设置，loop.run_until_complete(...) 运行，最后 loop.close() 关闭。
5. 用 async def 定义协程，在 I/O 等待处使用 await，让出控制权；对比普通函数，不要在协程中直接执行阻塞调用。
6. 用 asyncio.create_task(my_coroutine()) 调度协程并发运行，任务会立即开始；可先做其他工作，再 await task 获取结果。
7. 创建命名任务：asyncio.create_task(background_task(), name='my-background-task')，通过 task.get_name() 读取名称。
8. 用 asyncio.Future() 表示低层未来结果，通过 future.set_result(42) 设置结果，用 await future 获取结果；一般较少直接使用。
9. 用 asyncio.gather(...) 并发执行多个协程并等待全部完成，返回结果列表顺序与输入顺序一致；默认第一个异常会快速失败。
10. 用 asyncio.gather(..., return_exceptions=True) 收集异常而不是立即抛出，之后逐项用 isinstance(result, Exception) 判断失败。
11. 用 asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) 等待第一个完成；也可用 FIRST_EXCEPTION 或 ALL_COMPLETED；获得 done 和 pending 后，对 pending 任务调用 task.cancel()。
12. 用 asyncio.Semaphore(5) 限制最大并发数，并在 async with semaphore 内执行受限操作，例如限制 API 请求、数据库连接或外部服务调用。
13. 用 asyncio.Lock 保护共享资源临界区，在 async with lock 内读写共享状态，确保同一时间只有一个协程进入。
14. 用 asyncio.Event 协调多个协程：等待方 await event.wait()，触发方 event.set()，可让多个等待者同时被唤醒。
15. 用 asyncio.Queue 实现生产者/消费者：生产者 await queue.put(item)，消费者 await queue.get()，处理完成后 queue.task_done()。
16. 队列结束用 None 哨兵通知消费者；消费者收到 None 后 await queue.put(None) 传播给其他消费者并 break。
17. 取消任务时调用 task.cancel()；目标任务内捕获 asyncio.CancelledError 做资源清理，并 raise 重新抛出以标记取消完成；调用方 await task 时捕获 asyncio.CancelledError。
18. 超时取消：用 await asyncio.wait_for(task, timeout=5.0)；捕获 asyncio.TimeoutError 后调用 task.cancel()，再用 contextlib.suppress(asyncio.CancelledError) await task。
19. 获取任务异常：在任务完成后检查 task.done() and task.exception()；或在 asyncio.gather 中使用 return_exceptions=True。
20. 不要在每个请求中创建新的 aiohttp.ClientSession；复用一个 ClientSession，并用 asyncio.gather 并发执行多个 session.get(...)。
21. 使用 aiohttp.ClientSession 时，应在 async 函数内通过 async with aiohttp.ClientSession() as session 创建，以确保连接清理、防止资源泄漏、正确管理 SSL 和优雅关闭。
22. 用类实现异步上下文管理器：async def __aenter__ 做异步 setup 并返回 self，async def __aexit__ 做异步 cleanup，然后用 async with Resource() as resource 使用。
23. 或用 @asynccontextmanager 装饰 async 生成器：yield 前 setup，yield 值，finally 中 cleanup，然后用 async with async_resource(...) as value 使用。
24. 不要在协程中调用阻塞库，例如 requests.get；改用 aiohttp 等异步库；如果必须调用阻塞函数，使用 loop.run_in_executor。
25. 运行阻塞代码：loop = asyncio.get_running_loop(); result = await loop.run_in_executor(None, blocking_io)。
26. 自定义线程池执行阻塞代码：用 ThreadPoolExecutor(max_workers=4) 创建 executor，并用 asyncio.gather 配合多个 loop.run_in_executor(executor, blocking_io) 并发执行。
27. 安排事件循环回调：loop.call_soon(callback, 'immediate') 立即回调；loop.call_later(2, callback, 'delayed') 延迟回调；loop.call_at(loop.time() + 3, callback, 'scheduled') 定时回调。
28. 测量异步性能：用 time.perf_counter() 记录开始和结束时间，计算 elapsed；对多个操作分别计时并打印。
29. 避免顺序 await：不要写 for 循环逐个 await fetch_data(i)，而是 tasks = [fetch_data(i) for i in range(10)]，再 results = await asyncio.gather(*tasks)。
30. 需要控制并发时，在批量任务中为每个任务包一层 asyncio.Semaphore(5)，再用 asyncio.gather 执行，以在提高吞吐的同时限制并发。
31. aiohttp 使用完成后可加 await asyncio.sleep(0)，让底层连接有机会关闭，尤其配合 ClientSession 清理。
32. 类中不要用类变量保存 aiohttp.ClientSession；在 __init__ 中设 self.session = None，在 __aenter__ 创建 aiohttp.ClientSession()，在 __aexit__ 中 await self.session.close()。
33. 不要忘记 await：用 result = await fetch_data()，不要写 result = fetch_data()，否则只得到 coroutine 而未执行。
34. 不要在事件循环中使用 time.sleep(5)；改用 await asyncio.sleep(5)，否则会阻塞整个事件循环。
35. 任务取消时必须在 except asyncio.CancelledError 中清理资源，并 raise 重新抛出，避免后台任务无法正确结束。
36. 避免死锁：使用多个 asyncio.Lock 时，所有协程都按相同顺序获取锁，例如都先 lock1 再 lock2。
37. 优雅关闭应用：设置 should_exit 标志，用 loop.add_signal_handler 监听 signal.SIGTERM 和 signal.SIGINT；收到信号后置 should_exit = True。
38. 优雅关闭时取消所有任务：for task in self.tasks: task.cancel()；然后 await asyncio.gather(*self.tasks, return_exceptions=True) 等待取消完成。
39. 在 aiohttp web.Application 中用 app.cleanup_ctx.append(background_tasks) 管理后台任务；background_tasks 是异步生成器，启动时 asyncio.create_task，yield，退出时 cancel 并用 suppress(asyncio.CancelledError) await。
40. 实现重试逻辑：retry_with_backoff(coro_func, max_retries=3, base_delay=1.0, max_delay=60.0, *args, **kwargs)；循环 await coro_func；最后一次失败则 raise；否则 delay = min(base_delay * (2 ** attempt), max_delay)，await asyncio.sleep(delay)。
41. 实现断路器：CircuitBreaker 维护 CLOSED、OPEN、HALF_OPEN 状态，以及 failure_threshold、recovery_timeout、success_threshold；OPEN 时若超过 recovery_timeout 转 HALF_OPEN，否则抛异常。
42. 断路器成功时重置 failure_count；HALF_OPEN 下成功次数达到 success_threshold 转 CLOSED；失败时 failure_count 增加，达到 failure_threshold 转 OPEN 并记录 opened_at。
43. WebSocket 多事件源：在 handler 中 asyncio.create_task(read_subscription(ws, redis)) 后台读 Redis 并 ws.send_str(message)；同时 async for msg in ws 处理入站消息；finally 中 task.cancel()。
44. 测试异步代码：pytest 使用 @pytest.mark.asyncio 标记 async def test_...；或用 unittest，在 setUp 中 asyncio.new_event_loop() 和 asyncio.set_event_loop()，测试内 loop.run_until_complete，tearDown 中 loop.close()。
45. 调试异步代码：asyncio.run(main(), debug=True) 或 loop.set_debug(True)；设置 loop.slow_callback_duration = 0.1；用 logging.basicConfig(level=logging.DEBUG) 记录慢回调。
46. 文档化异步函数：docstring 写 Args、Returns、Raises、Example、Note，说明必须在 async 上下文调用、连接池自动处理。
47. 并发 HTTP 示例：定义 async def fetch(session, url) 内 async with session.get(url) as response 并 return await response.text()；在 main 中 async with aiohttp.ClientSession() as session，用 asyncio.gather(*[fetch(session, url) for url in urls]) 并发抓取。
48. 限速 API 客户端：RateLimitedClient 用 asyncio.Semaphore(rate_limit) 控制并发；__aenter__ 创建 aiohttp.ClientSession，__aexit__ 中 await self.session.close() 并 await asyncio.sleep(0)；fetch 在 async with self.semaphore 内请求；fetch_all 用 asyncio.gather(*tasks, return_exceptions=True)。
49. 数据库连接池：AsyncConnectionPool 用 asyncio.Queue(maxsize=size) 存连接；init 创建连接并 put；acquire 用 await pool.get()；release 用 await pool.put(conn)；execute 中 acquire，try 执行，finally release；close 取出并关闭所有连接。
50. 实时数据处理器：DataProcessor 用 asyncio.Queue；producer 循环 put item，最后 put None；consumer 循环 get，收到 None 时传播 None 并 break，否则处理并 queue.task_done()；monitor 用 create_task 周期打印统计；run 用 asyncio.gather 启动多个 producer 和 consumer，结束后 cancel monitor。
51. 异步文件 I/O：使用 aiofiles.open 的 async with 写文件或读文件；process_files 用 asyncio.gather(*[read_file(path) for path in file_paths]) 并发读取，再用 zip 处理统计行数、单词数和字符数。
52. 事件循环最佳实践：简单程序用 asyncio.run()；不要在事件循环外创建 ClientSession；完成时关闭 loop；不要在事件循环中调用阻塞函数。
53. 自定义事件循环时：loop = asyncio.new_event_loop(); asyncio.set_event_loop(loop); try: loop.run_until_complete(main()) finally: loop.close()。
54. 在协程内用 asyncio.get_running_loop() 获取当前运行中的事件循环，避免混用已废弃或非运行中的 loop 获取方式。
55. 使用 async with AsyncDatabaseConnection('localhost') as db 管理异步数据库连接：__aenter__ 中连接，__aexit__ 中关闭，query 前检查连接是否存在。
56. 使用 async with async_resource('database') as db 通过 @asynccontextmanager 管理自定义异步资源，确保 setup 和 cleanup 总是成对执行。

适用边界 · 注意事项

• CPU 密集型计算不应使用 asyncio 来提速，应改用多进程或其他并行方案。
• 没有异步 I/O、只是简单顺序同步逻辑时，引入 asyncio 会增加复杂度而没有收益。
• 必须使用阻塞库且无法放入 run_in_executor 的场景，不要直接在事件循环中调用阻塞函数。