你写的 Python 代码，是不是经常在"等"？
一、一个让你扎心的场景
假设你要从 10 个网站抓取数据，你的代码可能是这样的：
import requestsfor url in urls: data = requests.get(url) # 等它返回... process(data)
这段代码的问题是什么？
它在等。请求发出去后，CPU 啥也不干，就干等着网络响应回来。10 个请求，每个等 1 秒，总共 10 秒。
但如果用异步编程：
import httpximport asyncioasync def fetch(url): async with httpx.AsyncClient() as client: data = await client.get(url) process(data)async def main(): tasks = [fetch(url) for url in urls] await asyncio.gather(*tasks)asyncio.run(main())
10 个请求并发执行，总耗时 ≈ 最慢的那个（约 1 秒），而不是 10 秒。
差了 10 倍。这就是异步编程的魅力。
二、一句话说清什么是异步
同步：点外卖，干等着饭做好，啥也不干。 异步：点外卖，等饭的时候刷剧、打游戏、看书，饭好了去拿。
异步编程的核心思想：等的时候干别的。
Python 里的 async/await 就是干这个的。
三、从零理解 async/await
3.1 什么是协程？
协程（Coroutine）就是一个可以暂停执行的函数。
async def hello(): print("你好") await asyncio.sleep(1) # 暂停 1 秒，让出控制权 print("世界")
看到 async def 了吗？这定义了一个协程函数。
看到 await 了吗？这告诉 Python："这里要等，你先去干别的，等好了叫我。"
3.2 怎么运行协程？
import asyncioasync def say_hello(): print("Hello") await asyncio.sleep(1) print("World")# 正确方式asyncio.run(say_hello())# ❌ 错误方式（新手常犯）# say_hello() # 只会返回一个协程对象，不会执行！
asyncio.run() 是协程的入口，就像 if __name__ == "__main__" 一样，是启动一切的地方。
3.3 并发执行多个任务
import asyncioasync def task(name, seconds): print(f"任务 {name} 开始，需要 {seconds} 秒") await asyncio.sleep(seconds) print(f"任务 {name} 完成") return f"{name} 的结果"async def main(): # 并发执行三个任务 results = await asyncio.gather( task("A", 3), task("B", 1), task("C", 2) ) print(results) # ['A 的结果', 'B 的结果', 'C 的结果']asyncio.run(main())
输出：
任务 A 开始，需要 3 秒任务 B 开始，需要 1 秒任务 C 开始，需要 2 秒任务 B 完成任务 C 完成任务 A 完成['A 的结果', 'B 的结果', 'C 的结果']
注意： 三个任务几乎同时开始，总共耗时 ≈ 3 秒（最慢的那个），而不是 3+1+2=6 秒！
四、实战：并发爬取网页
import asyncioimport httpxURLS = [ "https://httpbin.org/delay/3", # 3 秒后返回 "https://httpbin.org/delay/1", # 1 秒后返回 "https://httpbin.org/delay/2", # 2 秒后返回]async def fetch_url(url): async with httpx.AsyncClient() as client: print(f"开始请求: {url}") resp = await client.get(url, timeout=10) print(f"完成请求: {url}，状态码: {resp.status_code}") return resp.status_codeasync def main(): tasks = [fetch_url(url) for url in URLS] results = await asyncio.gather(*tasks) print(f"全部完成: {results}")asyncio.run(main())
运行结果： 总共耗时约 3 秒（而不是 6 秒），3 个请求同时发出，同时等待。
五、进阶：控制并发数量
如果 URL 有 1000 个，全部并发会把服务器打死，还会被限流。需要限制并发数：
import asyncioimport httpxsemaphore = asyncio.Semaphore(10) # 最多 10 个并发async def fetch_url(url): async with semaphore: # 这里会排队 async with httpx.AsyncClient() as client: resp = await client.get(url, timeout=10) return resp.status_codeasync def main(): urls = [f"https://httpbin.org/delay/1" for _ in range(100)] tasks = [fetch_url(url) for url in urls] results = await asyncio.gather(*tasks) print(f"成功 {len(results)} 个请求")asyncio.run(main())
asyncio.Semaphore(10) 就像限流闸门，同一时间最多 10 个协程在执行。
六、新手最容易踩的 3 个坑
❌ 坑 1：在异步函数里调同步阻塞代码
async def bad(): import requests resp = requests.get("https://example.com") # ❌ 阻塞了全局！ return respasync def good(): import httpx async with httpx.AsyncClient() as client: resp = await client.get("https://example.com") # ✅ 不会阻塞 return resp
解决： 同步代码用 asyncio.to_thread() 放到线程池执行。
❌ 坑 2：忘了 await
async def my_func(): return 42async def main(): result = my_func() # ❌ 返回的是协程对象，不是 42 result = await my_func() # ✅ 这才是 42
❌ 坑 3：重复调用 asyncio.run()
import asyncioasync def hello(): print("Hello")asyncio.run(hello())asyncio.run(hello()) # ❌ 在同一个线程里第二次调用会报错
解决： 整个程序只调用一次 asyncio.run()。
七、什么时候该用异步？
| | |
| 网络爬虫 | | |
| Web API 服务 | | |
| 数据库查询 | | |
| 文件读写 | | |
| CPU 密集计算 | | |
| 科学计算 / NumPy | | |
一句话总结： I/O 密集型（网络、磁盘）→ 异步；CPU 密集型（计算、训练）→ 多进程。
八、2026 年异步生态速查
| | |
| | httpx |
| | FastAPI |
| | SQLAlchemy 2.0 async |
| | redis-py async |
| | aiofiles |
写在最后
异步编程不是 Python 的"高级黑魔法"，而是一个实实在在的提效工具。
如果你刚开始学，记住三点就够了：
1.async def 定义协程2.await 让出控制权且不阻塞3.asyncio.gather() 并发执行多个任务
今天就从换掉 requests.get() 开始吧！
