你还在用多线程搞并发？
Python 里写高并发程序，很多人第一反应是 threading。
线程确实能用，但有个大坑：Python 有 GIL（全局解释器锁），多线程在 CPU 密集型任务上基本是伪并发，只能跑满一个核。
那 IO 密集型呢？比如同时请求100个 API、爬100个网页——线程也能做，但上下文切换开销大，代码还啰嗦。
asyncio 的出现就是为了解决这个问题。 它在单线程里实现伪并发，用协程替代线程，开销小、代码简洁、性能高。
下面我用实际例子，带你从入门到实战。
一、asyncio 基础：协程是什么？
协程（Coroutine）是 asyncio 的核心。它像函数，但可以用 await 暂停，等 IO 完成后继续执行。
import asyncio
async def fetch_data(url):
print(f"开始请求 {url}")
await asyncio.sleep(1) # 模拟网络请求
print(f"{url} 完成")
return {"data": "hello"}
# 运行协程
result = asyncio.run(fetch_data("https://api.example.com"))
print(result)
关键点：
async def定义协程函数await暂停当前协程，等 IO 完成asyncio.run()是入口，创建事件循环并运行协程
await 的本质：不是"等待"，而是"让出控制权"。当前协程暂停，事件循环去跑其他协程，等 IO 完成后再回来继续。
二、并发请求：比多线程快10倍
这是 asyncio 最经典的使用场景。
串行 vs 并发
先看不用 asyncio 的做法——串行请求：
import time
import asyncio
urls = ["https://api.example.com/a", "https://api.example.com/b", "https://api.example.com/c"]
start = time.time()
for url in urls:
asyncio.run(fetch_data(url))
print(f"串行耗时: {time.time() - start:.2f}秒")
# 输出: 串行耗时: 3.01秒
3个请求各等1秒，总共3秒。
用 asyncio.gather 并发
async def fetch_all():
tasks = [fetch_data(url) for url in urls]
results = await asyncio.gather(*tasks)
return results
start = time.time()
asyncio.run(fetch_all())
print(f"并发耗时: {time.time() - start:.2f}秒")
# 输出: 并发耗时: 1.01秒
3个请求同时发起，总共只等1秒。快了3倍。
asyncio.gather() 的作用：把多个协程打包在一起并发执行，等全部完成后返回结果。
三、控制并发数：别让服务器被打爆
并发不是越多越好。同时发起1000个请求，服务器可能扛不住，你的 IP 也可能被封。
用 asyncio.Semaphore 控制并发数：
async def fetch_with_limit(sem, url):
async with sem: # 获取信号量，并发数限制
return await fetch_data(url)
async def fetch_limited():
sem = asyncio.Semaphore(10) # 最多同时10个请求
tasks = [fetch_with_limit(sem, url) for url in urls]
results = await asyncio.gather(*tasks)
return results
这样即使有100个 URL，也最多同时发10个请求。其他90个排队等信号量。
四、异步 I/O：处理文件和网络
异步读写文件
import aiofiles # 需要 pip install aiofiles
async def read_large_file(path):
async with aiofiles.open(path, 'r') as f:
content = await f.read()
return content
async def write_result(path, data):
async with aiofiles.open(path, 'w') as f:
await f.write(data)
注意：标准库的 open() 是同步的，不能用 await。要装 aiofiles。
异步 HTTP 请求
import aiohttp
async def fetch_with_session(session, url):
async with session.get(url) as resp:
return await resp.json()
async def main():
async with aiohttp.ClientSession() as session:
tasks = [fetch_with_session(session, url) for url in urls]
results = await asyncio.gather(*tasks)
return results
aiohttp 是 asyncio 生态里最成熟的 HTTP 客户端库，比 requests 快得多。
五、实战：写一个异步爬虫
下面是一个完整案例——用 asyncio 并发爬取多个页面标题：
import asyncio
import aiohttp
from bs4 import BeautifulSoup
URLS = [
"https://news.ycombinator.com",
"https://github.com/trending",
"https://stackoverflow.com",
]
async def fetch_page(session, url):
try:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
html = await resp.text()
soup = BeautifulSoup(html, 'html.parser')
title = soup.title.string if soup.title else "No title"
return url, title
except Exception as e:
return url, f"Error: {e}"
async def main():
async with aiohttp.ClientSession() as session:
tasks = [fetch_page(session, url) for url in URLS]
results = await asyncio.gather(*tasks)
for url, title in results:
print(f"{url} -> {title}")
asyncio.run(main())
运行结果：
https://news.ycombinator.com -> Hacker News
https://github.com/trending -> Currently trending repositories · GitHub
https://stackoverflow.com -> Stack Overflow - Where Developers Learn & Grow
3个页面并发请求，总共1秒左右搞定。串行可能需要5-6秒。
六、常见坑和解决方案
坑1：忘了 await
async def wrong():
fetch_data("https://example.com") # 忘记 await，协程不会执行
解决：所有协程调用前面都要加 await。
坑2：在同步代码里直接调 async 函数
def sync_func():
result = asyncio.run(fetch_data("https://example.com")) # 可以，但有坑
如果在已经运行的事件循环里再调 asyncio.run()，会报错：RuntimeError: This event loop is already running。
解决：用 asyncio.create_task() 在已有事件循环里创建任务：
async def main():
task = asyncio.create_task(fetch_data("https://example.com"))
result = await task
坑3：混用同步和异步代码
import requests
async def wrong():
resp = requests.get("https://example.com") # 同步阻塞，会卡住整个事件循环
解决：异步代码里只用异步库。网络请求用 aiohttp，HTTP 请求用 httpx（支持 async/sync 双模式）。
七、asyncio vs 多线程：怎么选？
| 场景 | 推荐方案 |
|---|---|
| 大量 IO 操作（网络请求、文件读写） | asyncio |
| CPU 密集型计算 | multiprocessing |
| 简单并发，代码简洁优先 | threading（配合 GIL 释放的库） |
| 需要异步 + 同步混合 | asyncio + ThreadPoolExecutor |
一句话：IO 密集型选 asyncio，CPU 密集型选 multiprocessing，简单场景选 threading。
八、总结
asyncio 是 Python 高并发编程的利器。掌握以下几点就够用了：
async def+await是基本语法asyncio.gather()并发执行多个协程asyncio.Semaphore()控制并发数- 网络请求用
aiohttp，文件操作用aiofiles - 千万别在异步代码里用同步阻塞调用
你现在的项目里，有没有适合用 asyncio 改造的场景？聊聊看。