很多人刚开始学 Python 时都会在某个阶段被asyncio这套异步 I/O 库搞得晕头转向。网上能查到大量教程但大多数要么讲得过于底层上来就扔给你事件循环、协程调度、Future 对象这些概念要么就是只给三个例子你照着抄完还是不知道什么时候该用、怎么排查问题。作为一个在爬虫、消息队列、API 网关服务里实际踩过不少坑的 Python 使用者我打算用这篇博文把 asyncio 从“是什么、为什么”讲到“怎么写、怎么踩坑”尽量让一个刚入门的读者也能真正理解并跑通自己的异步代码。先给一个结论asyncio是 Python 提供的异步 I/O 编程框架核心是帮你在单线程内处理大量并发的 I/O 操作比如网络请求、文件读写、数据库查询、子进程调用。它特别适合 I/O 密集型任务对 CPU 密集型任务基本没有帮助。如果你是写爬虫、写 API 服务、写消息消费者、写需要同时等一堆外部服务响应的脚本那 asyncio 几乎是你绕不开的工具。这个内容适合所有已经能写简单 Python 脚本、但还没搞懂async/await和事件循环的开发者或者已经在用线程池但总感觉并发上不去、资源浪费严重的人。读完之后你不仅会写异步代码还会明白那些写法背后的调度逻辑。1. asyncio 到底在解决什么问题想理解 asyncio先得理解阻塞。你的程序每一次发 HTTP 请求、读本地文件、连数据库、调用第三方 API本质上都是在等“别人”。等的过程里 CPU 其实什么事都没干但写同步代码时整个线程就被卡住了后面的代码只能排队。同步代码像是只有一个收银员的超市顾客 A 买了十分钟东西收银员就站那儿等十分钟后面所有人陪着等。线程池方案像是多开了几个收银员每个收银员独立服务一个顾客看起来并行处理了但每个收银员依然会花大量时间干等顾客掏钱。asyncio 的方案则是一个收银员同时服务一堆顾客A 说“我要买大米”收银员把 A 的购物车清单登记好转身去帮 B 结算等 A 把米搬过来了收银员再回来收钱。收银员全程没有闲着CPU 也是这样。1.1 I/O 密集型 vs CPU 密集型任务asyncio 不解决所有并发问题。它只解决 I/O 密集型的场景。I/O 密集型任务的特点是程序大部分时间在等待外部资源CPU 计算量非常少。比如网络爬虫等待服务器返回 HTML、脚本等待数据库 exec 返回结果、大量调用 REST API。对这类任务异步单线程往往比多线程更快因为线程切换和锁的开销都没有了。CPU 密集型任务是另一回事比如视频编码、大矩阵计算、复杂的加密解密、数据清洗里的循环重计算。这类任务真正吃的是 CPU 算力如果你把一个 CPU 密集任务丢进 async 函数里它反而会阻塞事件循环导致其他协程全部无法执行。这时候正确选择是多进程multiprocessing或者把计算密集部分交给run_in_executor里的进程池而不是 asyncio 本身。区分方法很简单如果任务耗时主要花在await一个外部调用上它是 I/O 密集适合 asyncio如果耗时主要花在纯 Python 计算循环里它是 CPU 密集不适合 asyncio。我见过不少人拿 asyncio 去跑分页解析 Excel 或者做大量正则替换结果发现比同步版本还慢就是因为走了错误的赛道。1.2 同步阻塞带来的真实痛苦举一个很常见的场景。你需要从 100 个不同的 API 拉取数据然后汇总入库。同步代码的写法是import requests def fetch_sync(): results [] for i in range(100): # 每个请求耗时约 1 秒这里会依次等待 resp requests.get(fhttps://api.example.com/item/{i}) results.append(resp.json()) return results如果每个请求耗时 1 秒100 个请求就是 100 秒。这 100 秒里程序绝大部分时间都在傻等网络响应CPU 占用率几乎为 0。换用多线程会好转但 100 个线程带来的上下文切换、GIL 限制、线程安全问题又会出现。import asyncio import httpx async def fetch_async(): async with httpx.AsyncClient() as client: tasks [] for i in range(100): tasks.append(client.get(fhttps://api.example.com/item/{i})) responses await asyncio.gather(*tasks) return [resp.json() for resp in responses]这段代码发 100 个请求时不是一个个等而是同时发出 100 个请求在不违反目标服务器限额的前提下总耗时能压缩到大约 1 秒多也就是“最慢那一个请求”的耗时。这就是 asyncio 的价值。2. async/await 核心机制与事件循环从语法上看asyncio 代码就多了async def和await两个关键字。但很多人第一次写就摔跟头就是因为没搞懂背后那套调度机制。下面把核心机制一块一块拆开讲。2.1 协程不是线程也不是普通的“语法糖”协程coroutine是一个可以暂停和恢复的函数。当你用async def定义一个函数时调用它并不会真正执行函数体而是返回一个协程对象。这个对象在被 await 或加入事件循环之前函数体一行代码都不会运行。async def say_hello(): print(hello) return 1 # 这样并不会打印 hello coro say_hello() print(type(coro)) # class coroutine # 必须 await 或加入事件循环这跟普通函数完全不同也是新手最常见的困惑来源。普通函数调用即执行而协程调用只是“把菜谱写好”真正做饭要等事件循环来叫它。线程是操作系统管理的每条线程有自己的栈和寄存器状态切换由内核调度切换成本高。协程则完全在用户态运行切换只发生在遇到await时。因为切换不涉及系统调用协程之间切换的开销可以低到微秒级所以单线程里同时跑数千个协程完全没问题。2.2 事件循环的调度逻辑await 什么时候让出事件循环是整个 asyncio 的发动机。它是一个无限循环不断做三件事拿到当前可执行的协程、执行它、看它是否让出执行权。当协程遇到await某个可能耗时的操作时它会向事件循环注册一个“我还在等等到了叫我”然后立即跳回事件循环让其他协程执行。import asyncio async def worker(name, delay): print(f{name} 开始) await asyncio.sleep(delay) print(f{name} 结束) async def main(): await asyncio.gather( worker(task-a, 2), worker(task-b, 1), ) asyncio.run(main())运行一下你会发现顺序是task-a 开始、task-b 开始、task-b 结束、task-a 结束。task-a先执行到await asyncio.sleep(2)时暂停事件循环接管让task-b开始执行。sleep 本身不是真的“卡住”而是告诉事件循环2 秒之后再来叫我。等待期间事件循环干了很多事。这个行为也可以解释为什么 asyncio 里不允许写太重的同步逻辑。如果你在某个协程里写了一个time.sleep(2)或者一段会运行 5 秒的计算循环事件循环是没法“抢走”执行权的因为事件循环本身也是在同一线程里。它必须等这个协程自己主动让出。只有当代码执行到await时才会切回事件循环。2.3 tasks、coroutines、futures 三者关系三个概念常被混用但理解它们能省很多调试时间Coroutine协程对象async def函数调用后得到的对象本身还不能独立调度必须等 await。Task任务包装协程并把它提交给事件循环调度的对象。一个 Task 对应一个被调度的协程。Future一个更低层的对象表示一种“未来可能完成的结果”。Task 是 Future 的子类但日常开发中你不一定直接碰 Future大多数时候你处理的是 Task。有一段代码很直观import asyncio async def simple(): return done async def main(): task asyncio.create_task(simple()) print(task) # Task pending ... result await task print(result)asyncio.create_task(coro)是显式地把协程注册到事件循环中。之后即便你不立即 await这个协程也已经开始调度执行了。注意协程必须被调度才会运行光创建一个协程对象没有任何并发效果。如果你在main()中连续创建了两百个 task 然后各自await它们就会交替执行。3. 日常开发最常用的 asyncio APIasyncio 的 API 数量不少但日常开发高频用到的其实就那几个。把它们彻底吃透比背一百个 API 更管用。3.1 gather、wait 和 as_completed 的区别并发执行多个协程时最常用的方法是asyncio.gather。它接收一批协程对象或 Task返回一个列表列表顺序与传入顺序一致即使某个协程后完成它在结果列表里也还是在对应位置。import asyncio async def fetch_url(name): await asyncio.sleep(1) return name.upper() async def main(): results await asyncio.gather( fetch_url(aaa), fetch_url(bbb), fetch_url(ccc), ) print(results) # [AAA, BBB, CCC]gather有一个容易被忽略的行为如果不传return_exceptionsFalse一旦某个协程抛异常它会立即把异常抛给 await 的一方同时其他协程并不会自动停止但 gather 的结果可能不会再返回。如果你不想因为一个失败导致整个 gather 被中断可以传return_exceptionsTrue异常会作为结果元素返回方便逐个判断。asyncio.wait和 gather 不一样它更强调等待集合并且支持FIRST_COMPLETED、FIRST_EXCEPTION、ALL_COMPLETED这些模式。比如你要“等这批任务里最早完成的那个”wait 更合适。gather 的返回结果不包含任务本身状态而 wait 返回(done, pending)两组任务集合自由度更高。asyncio.as_completed则是流式处理。它接收一批任务返回一个迭代器每完成一个任务就会 yield 一个结果所以你可以边等边处理而不是全跑完再统一拿结果。适合类似“100 个请求陆续完成每完成一个就立即解析和写入数据库”的场景。3.2 run_in_executor把同步代码变成异步asyncio 只能让自己的协程实现异步但它无法把任意一个普通同步函数变成非阻塞。比如你用requests.get()这种同步库发请求直接放在协程里就是阻塞的。解决办法是用loop.run_in_executor它能将同步函数放到线程池或进程池里执行从而不阻塞事件循环。import asyncio import requests def sync_request(url): return requests.get(url).status_code async def async_context(): loop asyncio.get_running_loop() # 使用默认线程池执行同步请求 result await loop.run_in_executor(None, sync_request, https://api.example.com/health) print(result)在 Python 3.9 之后的代码里官方推荐用asyncio.to_thread(func, *args)达到同样的效果它的写法更简洁。import asyncio import requests async def main(): result await asyncio.to_thread(requests.get, https://api.example.com/health) return result.status_code这里要注意run_in_executor可以传ThreadPoolExecutor或ProcessPoolExecutor实例。如果任务是 CPU 密集型请用ProcessPoolExecutor因为多线程在 GIL 下没法利用多核如果任务主要是 I/O 密集默认的线程池就够了。3.3 asyncio.Queue控制并发、传递数据当任务数量特别大时一次性创建几百个 Task 可行但创建几万、几十万个就有点危险了事件循环维护它们也会吃力。更合理的做法是用生产者消费者模式搭配asyncio.Queue控制并发和传递数据。队列一个有界对象可以设定maxsize。生产者往队列里put数据消费者从队列里get数据。队列的put和get都是异步方法在队列满时会自动让出执行权队列空时消费者也会等待而不是空转。import asyncio import random async def producer(queue, total): for i in range(total): await queue.put(i) print(f生产: {i}) await asyncio.sleep(0.2) # 发送终止信号 await queue.put(None) async def consumer(queue, name): while True: item await queue.get() if item is None: await queue.put(None) # 继续传递终止信号给下一个消费者 break print(f消费者 {name} 处理: {item}) await asyncio.sleep(random.uniform(0.1, 0.3)) async def main(): q asyncio.Queue(maxsize10) producers [asyncio.create_task(producer(q, 20))] consumers [asyncio.create_task(consumer(q, fc{i})) for i in range(3)] await asyncio.gather(*producers) await asyncio.gather(*consumers) asyncio.run(main())这里有个细节多个消费者同时从一个队列取数据时每个消息只会被消费一次asyncio.Queue 天然是线程安全的协程安全容器不需要额外加锁。队列的join()方法可以等待队列清空是生产者消费者模式中常用的同步手段但与简单 get 不同你需要同时task_done()。3.4 超时管理与取消协程真实场景里网络请求可能会卡住很久。如果不设超时协程可能一直挂着。大多数人会用asyncio.wait_forimport asyncio async def slow_work(): await asyncio.sleep(100) return done async def main(): try: result await asyncio.wait_for(slow_work(), timeout3) except asyncio.TimeoutError: print(超时了)注意这里的行为超时发生后被等待的协程内部会被取消也就是CancelledError会在await处被注入。如果协程捕获并吞掉了 CancelledErrorwait_for是没法真正终止它的任务会继续跑。这种情况被称为“任务泄漏”是异步服务里比较隐蔽的资源浪费。正确清理自己的协程需要用到try/finally。如果协程持有数据库连接或文件句柄在 finally 里释放是必须的操作。下面展示一个协作式取消的写法async def worker(): try: while True: await asyncio.sleep(1) print(工作中) except asyncio.CancelledError: print(收到取消通知) raise # 一定要重新抛出才算真正取消细节在于except asyncio.CancelledError里完成资源清理后要重新raise否则事件循环会认为任务已被取消但协程没有正确响应可能导致Task was destroyed but it is pending之类的警告。4. 实战异步请求调度怎么写才能稳定理论讲完来做一个相对完整的实战。目标是用 asyncio 实现一个“可控并发、有重试、有超时”的请求调度器。这个模式在爬虫、批量查询、报表回填任务里都能复用。4.1 选择异步 HTTP 客户端Python 里主流的异步 HTTP 客户端有两个aiohttp和httpx。 aiohttp 在 asyncio 生态里出生得更早功能完善支持 WebSocket、连接池但 API 风格略老。 httpx 提供同步和异步两套接口如果你已有requests使用经验上手 httpx 会平滑很多。我个人在写新项目时更倾向用 httpx因为它对 HTTP/2、代理、超时配置更友好而且和 requests API 相似不太容易踩惯性思维的坑。安装很简单pip install httpx4.2 一个可控并发的请求调度器先定义一个异步 fetch 函数包含超时和简单重试。我建议把“单次请求”封装成一个函数后面加并发、加信号量都容易扩展。import asyncio import httpx async def fetch_with_retry(client, url, retries3, timeout5.0): for attempt in range(retries): try: resp await client.get(url, timeouttimeout) if resp.status_code 200: return resp.text # 4xx 大多数不用重试5xx 和网络错误才需要重试 elif resp.status_code 500: return None except (httpx.TimeoutException, httpx.NetworkError) as e: if attempt retries - 1: print(f{url} 请求失败: {e}) return None await asyncio.sleep(0.5 * (attempt 1)) # 简单的线性退避 return None接着用asyncio.Semaphore控制并发数。因为一次性发起几千个并发请求很容易把目标服务打爆也可能被自己的操作系统文件描述符限制挡住。async def bounded_fetch_all(urls, max_concurrency10): semaphore asyncio.Semaphore(max_concurrency) async with httpx.AsyncClient() as client: async def one_task(url): async with semaphore: return url, await fetch_with_retry(client, url) results [] for task in asyncio.as_completed([one_task(url) for url in urls]): url, content await task results.append((url, content)) if content is None: print(f记录失败: {url}) return results async def main(): urls [fhttps://httpbin.org/delay/{i % 3} for i in range(50)] results await bounded_fetch_all(urls, max_concurrency8) print(f完成: {len(results)})这里asyncio.Semaphore的本质是限制同一时刻能进入临界区的协程数。当并发任务多时信号量内部会让超出的协程等待。把信号量的获取写在每个任务的入口处可以保证并发不会超过设定值。4.3 如何优雅处理失败和中间结果在实际任务里我不建议把所有结果都攒到内存里再处理数据量大时很容易 OOM。使用asyncio.as_completed或queue边取边存更合理。上面的例子用了 as_completed一旦某个请求完成立即可以把结果写到文件、数据库或传给下一个处理函数。如果你需要严格保持提交顺序比如批量更新第 i 条记录的响应要写回第 i 条的状态那么gather的一一对应结果列表更方便。两者没有绝对优劣按数据消费方式选择。还要注意异步 HTTP 客户端默认复用连接池。如果你在循环里反复创建AsyncClient连接不会被复用每次握手都是新的 TCP 连接性能会差很多。应该在一个会话生命周期内复用一个 client这也是上面代码把一个 client 作为参数传给 fetch 函数的原因。5. 异步调试与常见坑速查asyncio 项目跑起来之后经常出现“代码看起来没问题但就是卡死或疯狂报错”的情况。下面整理一些我实际踩过、也帮同事排查过的常见问题表格和注释都给了排查思路。5.1 在同步函数里调用 async 函数时报错很多人在 Flask/Django 的视图函数里直接async_func()发现没有任何输出或者干脆报RuntimeWarning: coroutine was never awaited。原因再强调一次调用 async 函数返回的是协程对象不会执行。在同步函数里想运行一个 async 函数最简单的做法是import asyncio def sync_func(): asyncio.run(some_async())但注意在已经有一个运行中事件循环的线程里再次调用asyncio.run()会报RuntimeError: asyncio.run() cannot be called from a running event loop。这时不要硬调而是把异步逻辑交给asyncio.run_coroutine_threadsafe或者在你的异步框架里单独开线程使用独立事件循环。5.2 有一个协程卡住其他全部不动事件循环被阻塞了十有八九是某个协程里使用了同步阻塞调用比如time.sleep()、requests.get()或者大计算循环。很多人打印日志时发现第一个任务跑完第二个任务迟迟不开始就以为是并发没生效实际是事件循环被第一个任务的同步操作卡死了。排查思路在任务日志里每隔一段时间打印当前所有任务状态用asyncio.all_tasks()可以看到事件循环里还存在哪些任务。import asyncio async def monitor(): while True: tasks [t for t in asyncio.all_tasks() if t is not asyncio.current_task()] for t in tasks: print(t, t._state) await asyncio.sleep(5)不过_state是私有属性不建议在产品代码里用。更稳妥的方式是把关键业务流程包一层带超时的asyncio.wait_for超时后直接打日志。5.3 事件循环选型run 还是 get_event_loopPython 3.10 之后官方对事件循环的管理做了清理。以前常见的loop asyncio.get_event_loop(); loop.run_until_complete(main())这种写法在新版本里容易碰到DeprecationWarning。 对绝大多数应用场景直接使用asyncio.run(main())asyncio.run()会自己创建事件循环、运行 main 协程最后关闭事件循环省去手动管理资源的问题。它可以多次调用每次调用都会新建一个循环。但要注意它不可以在一个已经运行的事件循环里嵌套调用。如果是在 Jupyter Notebook、既有 event loop 的服务里更推荐await main()或通过nest_asyncio打补丁非必要不用补丁会比较粗暴。5.4 多线程里同时跑 asyncio 的问题asyncio 是单线程的事件循环但有时你确实需要在一个多线程程序的子线程里各自跑一个异步任务。此时核心原则是同一时刻一个线程只允许有一个事件循环不同线程的事件循环不能互相交叉操作 Task。 错误示范是把线程 A 里创建的 Task 传给线程 B 等待这会导致不稳定。正确做法是让线程 A 只负责运行自己的事件循环线程间数据用queue.Queue或concurrent.futures传递。5.5 调试异步代码的工具建议没有哪个人能只看日志就解决所有异步问题。我最常用的工具组合如下asyncio.run(main(), debugTrue)开启调试模式会显示任务执行耗时、未 await 的协程、回调执行时长PYTHONASYNCIODEBUG1环境变量效果同上faulthandler程序卡住时发送SIGABRT打印线程栈能定位到底是哪一行没让出事件循环pytest-asyncio写单元测试时模拟异步逻辑的标准方案比自己在测试里手写asyncio.run更优雅。5.6 常用异常情况速查现象常见原因解决方案RuntimeWarning: coroutine was never awaited把 async 函数当作普通函数调用没有 await 或创建任务补上 await或用 asyncio.create_task 调度RuntimeError: no running event loop在未运行事件循环的同步线程里调用 get_event_loop 或 create_task改用 asyncio.run 包一层RuntimeError: This event loop is already running在已运行的 loop 内调用 asyncio.run / loop.run_forever直接 await 内部协程或换 run_coroutine_threadsafeTask was destroyed but it is pending任务没等它结束变量被回收或循环已关闭尽量 await 所有 task关闭时先取消未完成任务asyncio.TimeoutError 后协程仍在执行wait_for 超时只是取消尝试协程吞掉了取消在协程内捕获 CancelledError 后 re-raise异步 HTTP 请求偶尔连接卡死连接池复用问题或没有设置超时尽量全局复用一台 AsyncClient设置 timeout6. 扩展思考asyncio 与线程池搭配的正确姿势asyncio 不是银弹它在实际项目里经常和线程、进程池一起使用组合得当能发挥最大的威力。比如一段业务逻辑里既要调外部 HTTP API又要把返回结果交给本地一个 CPU 密集的解析函数最合理的结构是HTTP 部分用 asyncioCPU 密集部分用asyncio.to_thread或run_in_executor(ProcessPoolExecutor)执行避免阻塞事件循环。通过这种方式资源利用率通常会比单纯用线程池高不少也比“同步代码 大量线程”的并发模型更可控。我之前做过一个 RPA 任务调度器需要同时监控几百个业务流程的状态每个流程都要周期性地查数据库、调接口、等固定时间。如果给每个流程开一条线程几百条线程既是系统负担也让状态同步变得异常困难。后来把所有流程改成协程定时等待用asyncio.sleep数据库查询的同步客户端包进to_thread整体代码量更少、并发能力也上去了。对我个人来说这种“异步驱动、同步适配”的混合模式才是 asyncio 最值得使用的姿势。最后再分享一个实操里的小技巧新手可以从async/await httpx asyncio.gather这个最小组合开始练手先写一个能并发抓取 20 个网页的程序再把队列、信号量、重试一个个加进去。遇到卡住就开debugTrue跑一遍绝大多数问题都会自己暴露出来。踩过几次坑之后你再看 asyncio 就很容易形成直觉了。