为什么需要 asyncio？
假设你要写一个爬虫，下载 100 张图片。传统的同步写法会一张一张下载，每张耗时 1 秒，总耗时 100 秒。而使用多线程，虽然能并发，但线程切换开销大，且 GIL 限制了 CPU 密集型任务。asyncio 提供了另一种思路：单线程内通过事件循环调度协程，在 I/O 等待时自动切换任务，从而用极少的资源实现高并发。
但很多初学者对 asyncio 的理解停留在 async/await 语法层面，遇到 RuntimeError: This event loop is already running 或任务未执行就退出等问题时一头雾水。本文将从事件循环的本质讲起，带你手写一个迷你事件循环，再逐步构建一个异步爬虫，最后给出生产环境的最佳实践。
1. 核心概念：协程、任务与事件循环
1.1 协程（Coroutine）
协程是通过 async def 定义的函数，调用它不会立即执行，而是返回一个协程对象。要运行它，必须交给事件循环。
import asyncio
async def hello():
print("Hello")
await asyncio.sleep(1)
print("World")
# 直接调用 hello() 只会返回协程对象，不会执行
coro = hello()
print(coro) # <coroutine object hello at 0x...>
# 正确方式：asyncio.run 创建事件循环并运行
asyncio.run(hello())
1.2 任务（Task）
任务是对协程的封装，它会被事件循环调度执行。asyncio.create_task() 将协程包装为任务并立即排入事件循环。
async def main():
task1 = asyncio.create_task(hello())
task2 = asyncio.create_task(hello())
await task1
await task2
asyncio.run(main())
1.3 事件循环（Event Loop）
事件循环是 asyncio 的心脏，它负责：
- 注册任务和回调
- 在 I/O 就绪时调度对应任务
- 处理定时器
一个线程只能有一个运行中的事件循环。asyncio.run() 会创建新循环，运行完后关闭。
2. 手写迷你事件循环：理解调度原理
为了透彻理解，我们用生成器模拟一个极简事件循环。
import time
from collections import deque
class MiniEventLoop:
def __init__(self):
self._ready = deque() # 就绪任务队列
self._sleeping = [] # 睡眠中的任务 (唤醒时间, 任务)
def call_soon(self, callback, *args):
"""将回调加入就绪队列"""
self._ready.append((callback, args))
def call_later(self, delay, callback, *args):
"""延迟执行"""
self._sleeping.append((time.time() + delay, callback, args))
def run_forever(self):
while self._ready or self._sleeping:
# 处理已就绪的回调
while self._ready:
callback, args = self._ready.popleft()
callback(*args)
# 处理睡眠任务
if self._sleeping:
self._sleeping.sort(key=lambda x: x[0])
now = time.time()
while self._sleeping and self._sleeping[0][0] <= now:
_, callback, args = self._sleeping.pop(0)
callback(*args)
else:
# 没有就绪任务时，等待最近的任务
if self._sleeping and not self._ready:
sleep_time = self._sleeping[0][0] - time.time()
if sleep_time > 0:
time.sleep(sleep_time)
# 使用示例
def task1():
print("Task 1 start")
loop.call_later(1, task1_continue)
def task1_continue():
print("Task 1 continue after 1s")
def task2():
print("Task 2 start")
loop = MiniEventLoop()
loop.call_soon(task1)
loop.call_soon(task2)
loop.run_forever()
这个迷你循环虽然简单，但揭示了 asyncio 的核心：就绪队列 + 定时器管理。真正的 asyncio 使用 select/epoll/kqueue 监听 I/O 事件，并集成 Future 和协程驱动。
3. 深入 async/await 执行流程
await 表达式会挂起当前协程，将控制权交还给事件循环。当被等待的对象完成时，事件循环再恢复协程。
import asyncio
async def compute(x, y):
print(f"Compute {x} + {y} ...")
await asyncio.sleep(1) # 模拟 I/O 等待
return x + y
async def main():
# 并发运行两个协程
result = await asyncio.gather(compute(1, 2), compute(3, 4))
print(result) # [3, 7]
asyncio.run(main())
asyncio.gather 将多个协程包装为任务并发执行，并收集结果。注意：gather 返回的结果顺序与传入顺序一致，与完成顺序无关。
4. 实战：异步爬虫（aiohttp + asyncio）
下面我们构建一个异步爬虫，抓取多个网页的标题。使用 aiohttp 库。
4.1 安装依赖
pip install aiohttp beautifulsoup4
4.2 基础版本
import asyncio
import aiohttp
from bs4 import BeautifulSoup
async def fetch_title(session, url):
async with session.get(url) as response:
html = await response.text()
soup = BeautifulSoup(html, 'html.parser')
title = soup.title.string if soup.title else 'No title'
return url, title
async def main(urls):
async with aiohttp.ClientSession() as session:
tasks = [fetch_title(session, url) for url in urls]
results = await asyncio.gather(*tasks)
for url, title in results:
print(f"{url}: {title}")
if __name__ == '__main__':
urls = [
'https://www.python.org',
'https://www.github.com',
'https://www.stackoverflow.com'
]
asyncio.run(main(urls))
4.3 控制并发数
直接 gather 所有请求可能压垮目标服务器或触发限制。使用 asyncio.Semaphore 限制并发数。
async def fetch_title(session, url, sem):
async with sem:
async with session.get(url) as response:
html = await response.text()
soup = BeautifulSoup(html, 'html.parser')
title = soup.title.string if soup.title else 'No title'
return url, title
async def main(urls, concurrency=5):
sem = asyncio.Semaphore(concurrency)
async with aiohttp.ClientSession() as session:
tasks = [fetch_title(session, url, sem) for url in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)
for res in results:
if isinstance(res, Exception):
print(f"Error: {res}")
else:
print(f"{res[0]}: {res[1]}")
4.4 超时与重试
网络请求必须设置超时，并考虑重试。
import asyncio
from aiohttp import ClientTimeout
async def fetch_with_retry(session, url, sem, retries=3):
async with sem:
for attempt in range(retries):
try:
timeout = ClientTimeout(total=10)
async with session.get(url, timeout=timeout) as response:
html = await response.text()
soup = BeautifulSoup(html, 'html.parser')
title = soup.title.string if soup.title else 'No title'
return url, title
except Exception as e:
if attempt == retries - 1:
raise
await asyncio.sleep(1 * (attempt + 1)) # 指数退避
5. 常见陷阱与最佳实践
5.1 阻塞调用阻塞整个事件循环
在协程中调用同步阻塞函数（如 time.sleep、requests.get）会阻塞事件循环，导致所有任务卡住。
错误示例：
async def bad():
time.sleep(5) # 阻塞事件循环！
解决方案：使用 asyncio.tothread 或 runin_executor 将阻塞调用放到线程池。
async def good():
await asyncio.to_thread(time.sleep, 5) # 不阻塞事件循环
5.2 忘记 await 导致任务未执行
async def main():
asyncio.create_task(hello()) # 创建任务但未等待
# 如果 main 很快结束，任务可能未完成
asyncio.run(main())
修复：保存任务引用并 await，或使用 asyncio.gather。
5.3 在同步代码中调用异步函数
不能在普通函数中直接 await。必须通过 asyncio.run() 或 loop.rununtilcomplete() 启动事件循环。
5.4 使用 asyncio.Queue 实现生产者-消费者
import asyncio
import random
async def producer(queue):
for i in range(5):
await asyncio.sleep(0.5)
item = random.randint(1, 100)
await queue.put(item)
print(f"Produced {item}")
await queue.put(None) # 结束信号
async def consumer(queue):
while True:
item = await queue.get()
if item is None:
break
print(f"Consumed {item}")
await asyncio.sleep(1)
async def main():
queue = asyncio.Queue(maxsize=3)
await asyncio.gather(producer(queue), consumer(queue))
asyncio.run(main())
6. 性能对比：同步 vs 多线程 vs asyncio
我们模拟 100 个 I/O 任务（每个等待 0.1 秒），比较三种方式。
import asyncio
import time
import requests
from concurrent.futures import ThreadPoolExecutor
# 同步
start = time.time()
for _ in range(100):
time.sleep(0.1)
print(f"Sync: {time.time() - start:.2f}s") # ~10s
# 多线程
start = time.time()
with ThreadPoolExecutor(max_workers=10) as executor:
list(executor.map(lambda _: time.sleep(0.1), range(100)))
print(f"Thread: {time.time() - start:.2f}s") # ~1s
# asyncio
async def io_task():
await asyncio.sleep(0.1)
async def async_main():
await asyncio.gather(*(io_task() for _ in range(100)))
start = time.time()
asyncio.run(async_main())
print(f"Asyncio: {time.time() - start:.2f}s") # ~0.1s
结果：asyncio 远胜同步，且比多线程更轻量。但注意：asyncio 只适用于 I/O 密集型任务，CPU 密集型仍需多进程。
7. 总结
- asyncio 通过事件循环在单线程内实现并发，适合 I/O 密集型场景。
- 协程是核心，
await挂起协程，create_task调度任务。 - 避免阻塞事件循环，用
to_thread处理同步阻塞调用。 - 使用
Semaphore控制并发，gather收集结果，Queue实现生产者-消费者。 - 性能对比显示 asyncio 在高并发 I/O 下优势明显。
下一步可以学习 uvloop 提升事件循环性能，或研究 asyncio 与多进程结合处理 CPU 密集型任务。
延伸阅读：
- Python 官方文档 asyncio 章节
- 《Using Asyncio in Python》by Caleb Hattingh
- aiohttp 官方文档