开头部分异步编程这两年几乎成了 Python 开发者的必修课尤其是当你写爬虫、写接口、做数据处理发现程序老是卡在等待 IO 上时asyncio 就是那把能让你从“一个一个等”变成“同时等一堆”的钥匙。我最早接触 asyncio 是因为一个爬虫脚本要抓几万个页面用同步 requests 跑要俩小时后来换成 aiohttp asyncio 之后直接压缩到十几分钟那次优化给我留下极深的印象——从此只要碰上 IO 密集型任务我第一个想到的就是协程。这篇内容我会从 asyncio 最核心的事件循环、协程、任务、await 语法开始讲配合可以直接跑的代码示例带你从零搭起一个能用的异步程序。这篇内容适合谁已经会用 Python 写点简单脚本、但碰到 asyncio 代码就发懵的入门者也适合写过同步爬虫或脚本、想搞清楚 async/await 到底是怎么工作的同学。我会尽量不堆术语遇到绕不开的概念就用生活中的例子拆开讲保证你看完能自己动手写出第一个异步爬虫或异步请求程序。主体部分1. 异步编程到底解决了什么问题1.1 同步代码的痛点在哪里先看一个最常见的同步请求例子你用 requests 库抓 10 个网页程序会挨个发请求、挨个等响应。每个请求如果耗时 0.5 秒10 个请求就是 5 秒。这 5 秒里 CPU 大部分时间都在“干等”——数据还没回来代码只能卡在response requests.get(url)这一行什么都做不了。这就是同步阻塞模型的尴尬CPU 的运算能力被白白浪费在等待上。如果你抓的页面只有 10 个这点等待还能忍。可一旦页面数量上到几百、几千同步循环的耗时就是成倍增长而且这几乎是线性的。很多新手一开始没有意识到这一点脚本写到一半才发现速度慢得离谱然后开始查线程池、查并发最后绕了一大圈才找到 asyncio。我的建议是如果任务以网络请求、文件读写、数据库操作为主也就是所谓的 IO 密集型任务异步是最值得先试的方案代码写起来比多线程直观资源占用也更小。1.2 异步模型和同步模型的关键区别同步模型里函数调用是“一条道走到黑”的调用某个函数就必须等它返回才能继续往下走。异步模型则不一样它允许程序在遇到耗时操作时先“挂起”把控制权交还给事件循环然后去执行其他任务等之前的耗时操作完成了事件循环再回来接着往下跑。这种机制底层依赖的就是协程Python 里面用async def定义的函数就是一个协程函数调用它不会立即执行而是返回一个协程对象。很多人会问这不就是多线程吗区别其实挺大。多线程是靠操作系统调度线程来切换执行线程多了会有上下文切换开销还要考虑共享资源加锁的问题。协程是在单线程内靠事件循环手动切换任务切换成本极低而且因为同一时刻只有一个任务在跑代码普通变量不用加锁也不会有竞争问题特殊情况除外。这句话我再说白一点多线程像是多条流水线同时开工协程则是一条流水线里一个人同时管着好几道工序哪道工序要等待机器他就先去干别的。1.3 为什么 IO 密集型任务最适合 asyncio要理解这个得先明白 IO 密集型和 CPU 密集型的概念。IO 密集型任务的特点是大量时间花在“等”比如等网络响应、等磁盘读写、等数据库返回CPU 密集型任务则是一刻不停地在计算比如循环计算、复杂算法。asyncio 对 IO 密集型任务特别友好因为协程让等待时间重叠了——当一个请求在等网络另一个请求已经开始发了。几个请求的等待时间被叠加在一起总耗时就约等于最慢的那个请求而不是所有请求耗时之和。但如果你的任务是纯计算型的比如算斐波那契数列算到第 40 项那用 asyncio 是没用的因为整个处理过程 CPU 一直忙根本没有“等待间隙”可以利用。这种场景用多进程或者把计算扔给专用库是更合适的方向。刚开始学 asyncio 的人最容易犯的一个错就是在 CPU 密集任务上强行上异步最后发现速度反而更慢然后得出“asyncio 很垃圾”的错误结论。我想替 asyncio 说句话不是它不行是没用在它擅长的赛道上。2. 环境准备与第一个异步程序2.1 环境准备建议从 Python 3.7 开始asyncio 的 API 已经非常稳定目前最新版本更是顺手。我建议你用 Python 3.10 或更高版本来学因为 3.10 之后 asyncio 的报错信息更友好asyncio.run()这种高级入口用起来也最省心。先确认一下自己电脑上的环境打开终端运行python --version看看版本号。如果版本太低或者压根没装 Python那就先去官网下载一个最新版。装完 Python 之后建议顺手装一个虚拟环境特别是你手头有多个项目、依赖容易冲突的情况下。虚拟环境建起来很轻松python -m venv myenv # Windows 下激活 myenv\Scripts\activate # macOS / Linux 下激活 source myenv/bin/activate激活虚拟环境后你在里面装什么库都不会污染全局环境这一点我强烈建议从入门就开始养成习惯。后续示例中如果用到第三方库比如aiohttp就用pip install aiohttp直接装进当前环境。2.2 第一段可跑的 asyncio 代码光看理论没用直接上代码。先写一个最简单的异步函数感受一下async和await的基本用法import asyncio async def say_hello(): print(hello) await asyncio.sleep(1) print(world) asyncio.run(say_hello())这段代码的逻辑就是先打印 hello然后遇到await asyncio.sleep(1)挂起 1 秒1 秒后继续打印 world。注意你不能直接调用say_hello()因为它返回的是一个协程对象必须通过asyncio.run()或事件循环来运行。我又写了一个带并发效果的版本你可以对比一下两段代码执行时间的不同import asyncio import time async def task(name, delay): print(f{name} 开始) await asyncio.sleep(delay) print(f{name} 结束) async def main(): start time.time() await asyncio.gather( task(任务A, 2), task(任务B, 1), task(任务C, 3), ) print(f总耗时: {time.time() - start:.2f} 秒) asyncio.run(main())asyncio.gather()的作用是把多个协程打包并发执行。输出结果会显示 A 开始、B 开始、C 开始几乎同时发生结束时则是 B 先结束、A 其次、C 最后总耗时大约 3 秒而不是 6 秒。这个例子虽然简单但它就是异步并发的核心体验等待时间重叠了。2.3 事件循环是怎么“调度”这些任务的理解 asyncio绕不开“事件循环”这个概念。我常用一个餐厅例子来解释它事件循环就像一个餐厅服务员他手里有一堆订单任务。他接到一个订单后就把菜下到后厨发起耗时操作然后不等这道菜做完立刻去接其他订单、服务其他客人。后厨的菜做好了服务员再回来把菜端给对应客人。在这个例子里后厨就是操作系统底层的 IO 事件通知机制。协程执行到await时相当于服务员把工作挂起操作系统的 IO 完成之后会通知事件循环“那道菜好了”事件循环再把对应的协程重新拉回运行状态。整个过程没有多线程抢占全都是在一个线程里排队轮转所以切换开销极小。如果你想看看事件循环此刻到底在跑哪些任务可以用asyncio.all_tasks()获取当前所有任务列表这在调试并发问题时很有用async def main(): task1 asyncio.create_task(task(任务1, 2)) task2 asyncio.create_task(task(任务2, 1)) print(当前任务:, asyncio.all_tasks()) await task1 await task2从输出中你能直观地看到每个任务对象的 id 和状态这比瞎猜代码跑到哪一步要靠谱得多。3. 核心 API 与实操细节3.1 创建任务create_task vs gather vs ensure_future实际开发中很少直接await一个协程对象因为那样会让程序在等待时没有真正并行执行其他任务。更常见的写法是先把协程包装成任务Task再让事件循环调度它们。三种常见方式如下asyncio.create_task(coro)把协程包装成 Task立刻交给事件循环调度。注意它只能在事件循环运行时调用比如在async函数内部。asyncio.ensure_future(coro_or_future)功能类似 create_task但兼容性更好可以接受 Future 对象。在较老代码里比较常见现在官方推荐优先用 create_task。asyncio.gather(*coros, return_exceptionsFalse)批量并发执行多个协程并返回所有结果的有序列表。return_exceptionsTrue时即使某个任务异常也不会影响其他任务。我自己的习惯是如果用固定的一组协程优先asyncio.gather因为它能直接收集结果、写法最简洁。如果是动态生成、数量不定的任务比如从队列里不断取 URL 并发请求那就用create_task配合一个列表来管理。来看一个 gather 收集结果的例子import asyncio async def fetch_data(id): await asyncio.sleep(1) return f数据 {id} async def main(): results await asyncio.gather( fetch_data(1), fetch_data(2), fetch_data(3), ) print(results) asyncio.run(main())输出是[数据 1, 数据 2, 数据 3]顺序和传入顺序保持一致。这个特性在需要保持结果顺序的场景下非常好用。3.2 超时控制与任务取消异步程序里最怕的不是“慢”而是“永久卡住”。网络请求、外部接口随时可能抽风如果没有超时控制协程可能会挂在那里不动整个程序跟着倒霉。asyncio 提供了两个常用工具asyncio.wait_for()和asyncio.shield()。wait_for给协程设定一个最长等待时间超时就直接取消它import asyncio async def slow_task(): await asyncio.sleep(10) return 终于完成了 async def main(): try: result await asyncio.wait_for(slow_task(), timeout3) print(result) except asyncio.TimeoutError: print(任务超时被取消了) asyncio.run(main())这段代码里协程要睡 10 秒但只给它 3 秒时间超时后直接抛asyncio.TimeoutError。注意底层行为不只是抛异常它还会cancel()掉这个任务所以如果你对取消逻辑有特殊要求需要在任务内部捕获asyncio.CancelledError。shield的含义则是“保护”——它创建一个与内部协程关联的外层包装当外层任务被取消时内部协程不会跟着被取消。这在一些需要“即使超时了也不能中断底层任务”的场景很有用比如你要保存必要的数据、发送最后的上报请求等。我举一个典型例子某个后台任务正在把结果写进数据库如果因为调用者取消了任务而中断写入数据可能不完整这时候就可以用 shield 包住写入操作。3.3 控制并发数量Semaphore 信号量你可能会想既然 gather 这么爽那就一次并发 1000 个任务呗实际运行起来你大概率会碰到两种后果一种是目标服务器受不了直接拒绝连接另一种是自己的程序把文件描述符、内存撑爆。处理这种问题的常规方案是加一个并发上限用asyncio.Semaphore就能很好地控制。import asyncio sem asyncio.Semaphore(5) async def fetch_url(url): async with sem: print(f开始请求 {url}) await asyncio.sleep(1) print(f完成请求 {url}) async def main(): urls [fhttps://example.com/{i} for i in range(20)] await asyncio.gather(*[fetch_url(url) for url in urls]) asyncio.run(main())这段代码设置了同一时刻最多 5 个协程进入请求阶段剩下的 15 个协程都在门口排队等待。async with sem这种写法尤其直观进入代码块就占用一个信号量槽位退出就释放。我建议在写并发爬虫或批量请求脚本时不要把并发参数写死在代码里而是做成命令行参数或配置项方便随时调整。比如在本地测试用 2-3 个并发就够跑到服务器上再调大到 20-50这样能避免一次请求量过大把目标服务打挂。4. 从同步代码到异步的实战改造4.1 改造前一个同步请求版爬虫纸上谈兵没意思咱们直接把之前说的“同步爬虫变成异步爬虫”完整跑一遍。先看同步版本这个代码很简单用requests循环抓 10 个页面import requests import time URLS [ https://httpbin.org/delay/1 for _ in range(10) ] def fetch(url): resp requests.get(url, timeout5) return resp.status_code start time.time() for url in URLS: status fetch(url) print(f状态码: {status}) print(f同步版本总耗时: {time.time() - start:.2f} 秒)httpbin.org/delay/1是一个测试接口它在返回前强制会睡 1 秒用来模拟真实网络请求的延迟。同步版本跑完大约要 10 秒以上因为 10 个请求是一个接一个地等。4.2 改造后使用 asyncio aiohttp异步版本要用aiohttp这个支持异步的 HTTP 库先安装再替换请求部分pip install aiohttpimport asyncio import aiohttp import time URLS [ https://httpbin.org/delay/1 for _ in range(10) ] async def fetch(session, url): async with session.get(url, timeout5) as resp: return resp.status async def main(): async with aiohttp.ClientSession() as session: tasks [fetch(session, url) for url in URLS] results await asyncio.gather(*tasks) for status in results: print(f状态码: {status}) start time.time() asyncio.run(main()) print(f异步版本总耗时: {time.time() - start:.2f} 秒)异步版本跑完大约只需要 1-2 秒。注意一个细节ClientSession在并发请求前先创建一次然后把 session 传给每个 fetch 任务这样能复用连接池性能更好。不建议每个请求都单独创建一个 session那样连接管理反而是浪费。如果要加并发上限就把上面的sem asyncio.Semaphore(5)加进来在 fetch 里面用async with sem:包住请求部分。实际生产中这种“信号量 并发请求 超时控制”的组合几乎是无脑标配掌握了这一套大部分爬虫和批量请求场景都能应付。4.3 为什么性能提升这么明显有人会问“异步只是把等待时间重叠了但总请求数没变底层网速也没变为什么不慢反快”关键在于大多数机器在单个请求等待时根本没有用它最大的网络带宽。同步代码发一个请求后就傻等带宽空闲异步代码同时发起几十个请求把带宽利用起来了总体请求完成时间自然缩短。当然如果目标服务器带宽或你的运营商限制了总带宽并发再高也不会更快这时瓶颈就从“等待开销”变成了“网络带宽”。理解这一点就不会在优化时犯方向性错误。另外注意异步并发数量不是越大越好。我自己踩过一个坑用一个 200 并发的脚本去抓一个小网站结果对方直接把我的 IP 给封了。所以实用的做法是并发数从小往大调观察延迟和错误率再决定上限。对陌生网站我先用 5 试试确认没问题了再 20、50 一步步往上加。5. 常见问题与排查技巧实录5.1 报错This event loop is already running这个报错最常见于 Jupyter Notebook 或某些交互式环境。原因是这些环境本身已经启动了一个事件循环在跑这时候你在里面直接调用asyncio.run()就会撞车。解决方案要看场景如果你用的是 Jupyter直接await一个协程即可不需要asyncio.run()。如果你是写脚本先确认代码顶层没有别的事件循环。如果你确实需要在已运行的循环里再跑一个协程可以用asyncio.create_task()或者asyncio.get_running_loop().create_task()把它调度进去而不是再调用一次 run。这个报错对新手很不友好因为错误信息不够直白。我第一次在 Jupyter 里跑 asyncio 代码时也卡了半天最后才搞清楚是“循环套循环”的问题。5.2 为什么用了 asyncio 还是慢感觉像同步这种情况十有八九是因为协程里面调用了同步阻塞函数。比如常见错误写法是async def fetch_bad(url): resp requests.get(url) # 同步阻塞会卡住整个事件循环 return resp.text关键点在于requests.get()是同步阻塞的它在等待网络时不会让出事件循环所有并发协程都会在这被卡住最终效果跟同步代码一样。解决办法是换用异步库比如 aiohttp如果必须用同步库那就需要把阻塞调用丢到线程池里运行比如用await asyncio.to_thread(requests.get, url)。但这个算是不是特别优雅的办法优先更换库。我在代码评审时见过太多这种“假异步”代码了写法上看着有 async/await实际跑起来毫无并发效果。这里有个快速判断技巧看代码里 await 后面跟的是不是原生异步操作如果是asyncio.sleep、aiohttp.get、async with管理的内容基本没问题如果是requests.get、time.sleep、普通文件读取那多半是假异步。5.3 调试技巧开启 asyncio 的调试模式asyncio 自带一个调试模式专门用来抓两类问题协程没有及时被 await 导致的“嘟嘟响”警告以及任务回调执行时间过长。开启方式很简单asyncio.run(main(), debugTrue)或者在事件循环上直接设置loop asyncio.new_event_loop() loop.set_debug(True)开启调试模式后Python 会打印一些额外信息比如有哪些协程对象从未被 await哪些任务被创建后迟迟没有结果。日志里默认不会打印这些细节调试模式下则一目了然。有一次我在排查一个程序的“CPU 占用特别高”的问题就是用 debug 模式发现某个协程里有一个死循环式的密集计算阻塞了事件循环的正常轮转修完之后整体流畅了很多。5.4 常见问题速查表问题现象可能原因解决方案RuntimeError: asyncio.run() cannot be called from a running event loop事件循环已经在运行Jupyter 直接用 await脚本中避免重复调用 run并发任务执行后效果等同同步协程内调用了同步阻塞操作换用异步库或用asyncio.to_thread()程序抛出TimeoutErrorwait_for超时调整超时时间检查网络或目标服务响应警告coroutine was never awaited定义了协程但没 await 或 create_task检查协程调用是否漏加 await内存占用过高并发任务无限增长用 Semaphore 控制并发上限程序提前退出但还有任务没跑asyncio.run()在 main 返回后取消剩余任务在 main 中确保 gather 等待所有任务完成这张表里的前两个问题是新手最常遇到的强烈建议收藏一下等实际报错了再回来对照。6. 进阶方向与经验之谈6.1 asyncio 在真实项目里的生态位置只学 asyncio 本身还不够得看看它在生产环境长什么样子。以 Python 最火的几个方向为例FastAPI 的异步接口直接把 asyncio 带火了你用async def定义路径操作函数FastAPI 自动把它放进事件循环里跑。爬虫框架如 Scrapy 支持协程某些场景下能明显提升抓取效率aiohttp、httpx 则是日常自己写脚本时最常用的异步客户端。消息队列消费、WebSocket 长连接、定时任务调度这些场景也大量依赖 asyncio 的事件循环机制。也就是说学会 asyncio 不只是多了一个语法框架而是打开了异步编程的整个抽屉。后面你学 FastAPI、写并发脚本、做实时数据处理都会不断地碰到这套概念。6.2 我的几条实践建议第一新手不要在项目一开始就追求复杂的并发结构。先用 gather 把简单的并发跑通遇到问题再逐步加上信号量、超时、取消这些控制手段。一上来就搞一个庞大的任务调度系统大概率自己先被搞晕。第二理解 async/await 的最好方式是多调试、多打印。把事件循环开启时的顺序、每个任务的开始结束时间都打出来看多了你就知道协程是怎么切换的。这比反复看文档更有效。第三同步代码和异步代码不要混着写。当你用 asyncio 时尽量全部走异步库比如文件操作用aiofiles、数据库操作用对应的异步驱动。偶尔混一两个同步阻塞调用看似没事但在高并发量下会成为隐藏的性能瓶颈。第四测试异步代码时优先用pytest-asyncio这类插件。它允许你把测试函数直接标成异步然后由框架管理事件循环。十年前很多人因为不知道怎么写异步单测而放弃 asyncio现在这个门槛已经被磨平了。第五多读标准库源码。asyncio 的源码本身写得很清晰尤其是tasks.py里 Task 的状态转移逻辑读完你会对“任务是怎么被挂起和恢复的”形成深入理解。当然这一条放到入门的后期再拿前期先能跑通代码最重要。最后再分享一个我常用的套路写任何网络类异步脚本永远在入口函数外面包一层带try/finally的逻辑确保所有资源正常关闭、所有任务正常结束。比如用try: await gather包住主要逻辑finally: await session.close()。别小看这个习惯它能帮你避开很多“脚本跑完但进程还在挂起”的诡异问题。