其他
Python asyncio 并发 I/O…
用 asyncio 的协程、任务、同步原语和异步上下文管理器,把大量 I/O 密集型操作安全地并发化,并加入限流、取消、重试、断路与优雅关闭。
🧪 尚未被验证(还没有小呱复用过的记录,不代表不好)
GDI C 49/100
进化评分(GDI)
综合排序依据 —— 四个维度的加权结果,让你看出「为什么这条排在前面」
内在质量 85/100 ×0.35
使用度 0/100 ×0.30
口碑 50/100 ×0.20
新鲜度 59/100 ×0.15
适用场景
Python asyncio 并发 I/通用
解决思路
- 采用单线程协作式多任务模型编写程序:一个事件循环调度多个任务,任务只在 await 处主动让出控制权。
- 使用非阻塞 I/O 和 I/O 多路复用同时处理多个连接,避免在等待某个 I/O 时阻塞其他任务。
- 使用 asyncio.run(my_coroutine()) 运行顶层协程;Python 3.7+ 优先用这种方式。
- 需要手动管理事件循环时,使用 asyncio.get_event_loop() 或 asyncio.new_event_loop() 获取/创建循环,asyncio.set_event_loop(loop) 设置,loop.run_until_complete(...) 运行,最后 loop.close() 关闭。
- 用 async def 定义协程,在 I/O 等待处使用 await,让出控制权;对比普通函数,不要在协程中直接执行阻塞调用。
- 用 asyncio.create_task(my_coroutine()) 调度协程并发运行,任务会立即开始;可先做其他工作,再 await task 获取结果。
- 创建命名任务:asyncio.create_task(background_task(), name='my-background-task'),通过 task.get_name() 读取名称。
- 用 asyncio.Future() 表示低层未来结果,通过 future.set_result(42) 设置结果,用 await future 获取结果;一般较少直接使用。
- 用 asyncio.gather(...) 并发执行多个协程并等待全部完成,返回结果列表顺序与输入顺序一致;默认第一个异常会快速失败。
- 用 asyncio.gather(..., return_exceptions=True) 收集异常而不是立即抛出,之后逐项用 isinstance(result, Exception) 判断失败。
- 用 asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) 等待第一个完成;也可用 FIRST_EXCEPTION 或 ALL_COMPLETED;获得 done 和 pending 后,对 pending 任务调用 task.cancel()。
- 用 asyncio.Semaphore(5) 限制最大并发数,并在 async with semaphore 内执行受限操作,例如限制 API 请求、数据库连接或外部服务调用。
- 用 asyncio.Lock 保护共享资源临界区,在 async with lock 内读写共享状态,确保同一时间只有一个协程进入。
- 用 asyncio.Event 协调多个协程:等待方 await event.wait(),触发方 event.set(),可让多个等待者同时被唤醒。
- 用 asyncio.Queue 实现生产者/消费者:生产者 await queue.put(item),消费者 await queue.get(),处理完成后 queue.task_done()。
- 队列结束用 None 哨兵通知消费者;消费者收到 None 后 await queue.put(None) 传播给其他消费者并 break。
- 取消任务时调用 task.cancel();目标任务内捕获 asyncio.CancelledError 做资源清理,并 raise 重新抛出以标记取消完成;调用方 await task 时捕获 asyncio.CancelledError。
- 超时取消:用 await asyncio.wait_for(task, timeout=5.0);捕获 asyncio.TimeoutError 后调用 task.cancel(),再用 contextlib.suppress(asyncio.CancelledError) await task。
- 获取任务异常:在任务完成后检查 task.done() and task.exception();或在 asyncio.gather 中使用 return_exceptions=True。
- 不要在每个请求中创建新的 aiohttp.ClientSession;复用一个 ClientSession,并用 asyncio.gather 并发执行多个 session.get(...)。
- 使用 aiohttp.ClientSession 时,应在 async 函数内通过 async with aiohttp.ClientSession() as session 创建,以确保连接清理、防止资源泄漏、正确管理 SSL 和优雅关闭。
- 用类实现异步上下文管理器:async def __aenter__ 做异步 setup 并返回 self,async def __aexit__ 做异步 cleanup,然后用 async with Resource() as resource 使用。
- 或用 @asynccontextmanager 装饰 async 生成器:yield 前 setup,yield 值,finally 中 cleanup,然后用 async with async_resource(...) as value 使用。
- 不要在协程中调用阻塞库,例如 requests.get;改用 aiohttp 等异步库;如果必须调用阻塞函数,使用 loop.run_in_executor。
- 运行阻塞代码:loop = asyncio.get_running_loop(); result = await loop.run_in_executor(None, blocking_io)。
- 自定义线程池执行阻塞代码:用 ThreadPoolExecutor(max_workers=4) 创建 executor,并用 asyncio.gather 配合多个 loop.run_in_executor(executor, blocking_io) 并发执行。
- 安排事件循环回调:loop.call_soon(callback, 'immediate') 立即回调;loop.call_later(2, callback, 'delayed') 延迟回调;loop.call_at(loop.time() + 3, callback, 'scheduled') 定时回调。
- 测量异步性能:用 time.perf_counter() 记录开始和结束时间,计算 elapsed;对多个操作分别计时并打印。
- 避免顺序 await:不要写 for 循环逐个 await fetch_data(i),而是 tasks = [fetch_data(i) for i in range(10)],再 results = await asyncio.gather(*tasks)。
- 需要控制并发时,在批量任务中为每个任务包一层 asyncio.Semaphore(5),再用 asyncio.gather 执行,以在提高吞吐的同时限制并发。
- aiohttp 使用完成后可加 await asyncio.sleep(0),让底层连接有机会关闭,尤其配合 ClientSession 清理。
- 类中不要用类变量保存 aiohttp.ClientSession;在 __init__ 中设 self.session = None,在 __aenter__ 创建 aiohttp.ClientSession(),在 __aexit__ 中 await self.session.close()。
- 不要忘记 await:用 result = await fetch_data(),不要写 result = fetch_data(),否则只得到 coroutine 而未执行。
- 不要在事件循环中使用 time.sleep(5);改用 await asyncio.sleep(5),否则会阻塞整个事件循环。
- 任务取消时必须在 except asyncio.CancelledError 中清理资源,并 raise 重新抛出,避免后台任务无法正确结束。
- 避免死锁:使用多个 asyncio.Lock 时,所有协程都按相同顺序获取锁,例如都先 lock1 再 lock2。
- 优雅关闭应用:设置 should_exit 标志,用 loop.add_signal_handler 监听 signal.SIGTERM 和 signal.SIGINT;收到信号后置 should_exit = True。
- 优雅关闭时取消所有任务:for task in self.tasks: task.cancel();然后 await asyncio.gather(*self.tasks, return_exceptions=True) 等待取消完成。
- 在 aiohttp web.Application 中用 app.cleanup_ctx.append(background_tasks) 管理后台任务;background_tasks 是异步生成器,启动时 asyncio.create_task,yield,退出时 cancel 并用 suppress(asyncio.CancelledError) await。
- 实现重试逻辑: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)。
- 实现断路器:CircuitBreaker 维护 CLOSED、OPEN、HALF_OPEN 状态,以及 failure_threshold、recovery_timeout、success_threshold;OPEN 时若超过 recovery_timeout 转 HALF_OPEN,否则抛异常。
- 断路器成功时重置 failure_count;HALF_OPEN 下成功次数达到 success_threshold 转 CLOSED;失败时 failure_count 增加,达到 failure_threshold 转 OPEN 并记录 opened_at。
- WebSocket 多事件源:在 handler 中 asyncio.create_task(read_subscription(ws, redis)) 后台读 Redis 并 ws.send_str(message);同时 async for msg in ws 处理入站消息;finally 中 task.cancel()。
- 测试异步代码:pytest 使用 @pytest.mark.asyncio 标记 async def test_...;或用 unittest,在 setUp 中 asyncio.new_event_loop() 和 asyncio.set_event_loop(),测试内 loop.run_until_complete,tearDown 中 loop.close()。
- 调试异步代码:asyncio.run(main(), debug=True) 或 loop.set_debug(True);设置 loop.slow_callback_duration = 0.1;用 logging.basicConfig(level=logging.DEBUG) 记录慢回调。
- 文档化异步函数:docstring 写 Args、Returns、Raises、Example、Note,说明必须在 async 上下文调用、连接池自动处理。
- 并发 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]) 并发抓取。
- 限速 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)。
- 数据库连接池:AsyncConnectionPool 用 asyncio.Queue(maxsize=size) 存连接;init 创建连接并 put;acquire 用 await pool.get();release 用 await pool.put(conn);execute 中 acquire,try 执行,finally release;close 取出并关闭所有连接。
- 实时数据处理器: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。
- 异步文件 I/O:使用 aiofiles.open 的 async with 写文件或读文件;process_files 用 asyncio.gather(*[read_file(path) for path in file_paths]) 并发读取,再用 zip 处理统计行数、单词数和字符数。
- 事件循环最佳实践:简单程序用 asyncio.run();不要在事件循环外创建 ClientSession;完成时关闭 loop;不要在事件循环中调用阻塞函数。
- 自定义事件循环时:loop = asyncio.new_event_loop(); asyncio.set_event_loop(loop); try: loop.run_until_complete(main()) finally: loop.close()。
- 在协程内用 asyncio.get_running_loop() 获取当前运行中的事件循环,避免混用已废弃或非运行中的 loop 获取方式。
- 使用 async with AsyncDatabaseConnection('localhost') as db 管理异步数据库连接:__aenter__ 中连接,__aexit__ 中关闭,query 前检查连接是否存在。
- 使用 async with async_resource('database') as db 通过 @asynccontextmanager 管理自定义异步资源,确保 setup 和 cleanup 总是成对执行。
适用边界 · 注意事项
- CPU 密集型计算不应使用 asyncio 来提速,应改用多进程或其他并行方案。
- 没有异步 I/O、只是简单顺序同步逻辑时,引入 asyncio 会增加复杂度而没有收益。
- 必须使用阻塞库且无法放入 run_in_executor 的场景,不要直接在事件循环中调用阻塞函数。