异步生成器与异步迭代器
在协程函数中使用 yield，就变成了异步生成器。它每次 yield 产出值时都可以 await 其他协程。
async def async_generator():
for i in range(3):
await asyncio.sleep(1)
yield i
async def main():
# 必须用 async for 来遍历异步生成器
async for item in async_generator():
print(item) # 每隔 1 秒打印一个数字
asyncio.run(main())同样可以自定义**异步迭代器**，实现 `__aiter__` 和 `__anext__` 两个协议方法：
class AsyncCounter:
def __init__(self, limit: int):
self._limit = limit
self._count = 0
def __aiter__(self):
return self
async def __anext__(self) -> int:
self._count += 1
if self._count > self._limit:
raise StopAsyncIteration
await asyncio.sleep(0.5)
return self._count
async def main():
async for num in AsyncCounter(5):
print(num) # 每隔 0.5 秒打印 1 2 3 4 5对比同步迭代协议：
| 同步 | 异步 |
|---|---|
__iter__ / __next__ | __aiter__ / __anext__ |
for x in iterable | async for x in async_iterable |
StopIteration | StopAsyncIteration |
生成器 yield | 异步生成器 async def + yield |
异步上下文管理器
async with 背后是 __aenter__ 和 __aexit__ 两个协议方法，和同步的 with 类似，只是都是异步的。
class AsyncResource:
async def __aenter__(self):
print("获取资源")
await asyncio.sleep(0.5)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
print("释放资源")
await asyncio.sleep(0.5)
async def main():
async with AsyncResource() as res:
print("使用资源")对比同步上下文管理器：
| 同步 | 异步 |
|---|---|
__enter__ / __exit__ | __aenter__ / __aexit__ |
with ctx | async with ctx |
官方 asyncio 核心 API
sleep —— 非阻塞等待
使用 loop.call_later + Future 实现的延时：
def async_delay(duration: int):
loop = asyncio.get_event_loop()
future = loop.create_future()
loop.call_later(duration, future.set_result, None)
return future官方直接提供了 asyncio.sleep，用法完全相同：
async def main():
print("开始")
await asyncio.sleep(1) # 挂起当前协程 1 秒，事件循环去调度其他协程
print("1秒后")创建与运行协程
import asyncio
async def say_hello():
await asyncio.sleep(1)
return "Hello"
async def main():
# 1. asyncio.run —— 最高层入口
pass
result = asyncio.run(say_hello())create_task —— 将协程包装成 Task 并调度执行
async def main():
# 创建 Task，协程会立即被调度到事件循环中执行
task = asyncio.create_task(say_hello())
# 这里可以做别的事，task 已经在后台运行
result = await task # 等待 task 完成
print(result)gather —— 并发执行多个协程
async def fetch(url: str, delay: int) -> str:
await asyncio.sleep(delay)
return f"{url} 完成"
async def main():
# 同时发起多个请求，等待所有完成
results = await asyncio.gather(
fetch("url1", 2),
fetch("url2", 1),
fetch("url3", 3),
)
print(results) # ['url1 完成', 'url2 完成', 'url3 完成']
asyncio.run(main())gather 的特点：
- 所有协程并发执行
- 返回结果顺序与传入顺序一致
- 任何一个协程抛出异常，gather 会立即传播异常（其他协程仍会继续运行）
- 可通过
return_exceptions=True让异常以结果形式返回，不中断 gather
async def fail() -> str:
raise ValueError("出错了")
async def main():
results = await asyncio.gather(
fetch("ok", 1),
fail(),
return_exceptions=True, # 将异常作为返回值，不抛出
)
print(results) # ['ok 完成', ValueError('出错了')]wait —— 更灵活的等待方式
import asyncio
from typing import Coroutine
async def main():
tasks = [
asyncio.create_task(fetch("A", 2)),
asyncio.create_task(fetch("B", 1)),
asyncio.create_task(fetch("C", 3)),
]
# FIRST_COMPLETED: 任一完成就返回
# FIRST_EXCEPTION: 任一异常就返回
# ALL_COMPLETED: 全部完成（默认）
done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
print(f"已完成: {len(done)}, 待完成: {len(pending)}")
# 还可以手动处理未完成的任务
for task in pending:
task.cancel()as_completed —— 谁先完成谁先处理
async def main():
coros = [fetch("A", 2), fetch("B", 1), fetch("C", 3)]
for coro in asyncio.as_completed(coros):
result = await coro
print(result) # 按照完成的先后顺序打印TaskGroup —— 结构化并发（Python 3.11+）
async def main():
# TaskGroup 保证所有子任务在退出前完成
# 如果某个子任务异常，会取消组内所有其他任务
async with asyncio.TaskGroup() as tg:
task1 = tg.create_task(fetch("A", 2))
task2 = tg.create_task(fetch("B", 1))
task3 = tg.create_task(fetch("C", 3))
# 到这里所有任务都已安全完成
print(task1.result(), task2.result(), task3.result())Lock —— 互斥锁
多个协程可能竞争共享资源，Lock 保证同一时刻只有一个协程能访问
import asyncio
shared_data: int = 0
lock = asyncio.Lock()
async def safe_increment():
global shared_data
async with lock: # 获取锁，等锁释放前其他协程会在此等待
temp = shared_data
await asyncio.sleep(0) # 模拟耗时操作，此时切换协程也不会出问题
shared_data = temp + 1
async def main():
await asyncio.gather(*[safe_increment() for _ in range(100)])
print(shared_data) # 100Event —— 事件通知
一个协程等待另一个协程发出信号
async def waiter(event: asyncio.Event):
print("waiter: 开始等待")
await event.wait() # 等待事件被设置
print("waiter: 被唤醒")
async def setter(event: asyncio.Event):
print("setter: 1秒后设置事件")
await asyncio.sleep(1)
event.set() # 设置事件，唤醒所有等待者
async def main():
event = asyncio.Event()
await asyncio.gather(waiter(event), setter(event))waiter: 开始等待
setter: 1秒后设置事件
waiter: 被唤醒Semaphore —— 限制并发数
semaphore = asyncio.Semaphore(3) # 同时最多 3 个
async def limited_fetch(url: str):
async with semaphore: # 超过并发限制时等待
print(f"开始请求 {url}")
await asyncio.sleep(1)
print(f"完成请求 {url}")
return url
async def main():
urls = [f"url{i}" for i in range(10)]
await asyncio.gather(*[limited_fetch(url) for url in urls])Queue —— 异步队列
生产者-消费者模式的基石
import random
async def producer(queue: asyncio.Queue):
for i in range(10):
item = f"item_{i}"
await queue.put(item)
print(f"生产: {item}")
await asyncio.sleep(random.random())
await queue.put(None) # 发送结束信号
async def consumer(name: str, queue: asyncio.Queue):
while True:
item = await queue.get()
if item is None: # 收到结束信号
queue.task_done()
break
print(f"{name} 消费: {item}")
queue.task_done()
async def main():
queue = asyncio.Queue(maxsize=5)
await asyncio.gather(
producer(queue),
consumer("C1", queue),
consumer("C2", queue),
)asyncio.wait_for
异步编程中用于控制任务执行时间的核心函数。它允许你等待一个协程或任务（Task）完成，但前提是必须在指定的超时时间（timeout）内完成。
- 超时取消：如果在设定的
timeout秒数内，等待的对象（awaitable）没有执行完毕，wait_for会主动取消该任务，并抛出asyncio.TimeoutError异常。 - 正常返回：如果任务在超时前顺利完成，
wait_for会返回该任务的执行结果。 - 异常传播：如果等待的任务在执行过程中抛出了其他未处理的异常，该异常会直接传播给调用者。
async def slow_operation():
await asyncio.sleep(10)
return "完成"
async def main():
try:
result = await asyncio.wait_for(slow_operation(), timeout=2)
except TimeoutError:
print("操作超时了")asyncio.timeout（Python 3.11+）
async def main():
try:
async with asyncio.timeout(2):
result = await slow_operation()
except TimeoutError:
print("操作超时了")asyncio.timeout 是 Python 3.11 引入的现代超时控制机制，作为上下文管理器（Context Manager）提供。相比早期的 asyncio.wait_for()，它提供了更优雅、更灵活的任务超时管理方式。
在异步中运行同步代码
import time
def blocking_io() -> str:
time.sleep(0.5) # 同步阻塞操作
return "文件读取完成"
def cpu_intensive() -> int:
return sum(i * i for i in range(10_000_000))
async def main():
# to_thread：将同步阻塞函数放到线程池中执行
result = await asyncio.to_thread(blocking_io)
print(result)
# run_in_executor：更底层，可以指定执行器
loop = asyncio.get_running_loop()
result = await loop.run_in_executor(None, cpu_intensive)
print(result)1. IO 密集型同步代码（如网络请求、文件读写）
绝对不要直接在协程中调用（例如直接使用 requests.get() 或 time.sleep()），这会阻塞当前线程的事件循环。
解决方案：将同步函数扔到线程池中执行
现代写法（Python 3.9+）：使用 asyncio.to_thread()。它本质上是 run_in_executor 的简化版，无需手动获取事件循环，一行代码即可将同步函数包装为可 await 的协程。
import asyncio
import requests
async def fetch_data():
# 将同步的 requests.get 扔给默认线程池执行
response = await asyncio.to_thread(requests.get, 'http://example.com')
return response.status_code传统写法（兼容老版本）：使用 loop.run_in_executor(None, sync_func, args)。
loop = asyncio.get_running_loop()
result = await loop.run_in_executor(None, requests.get, 'http://example.com')第三方异步库
网络请求
aiohttp（第三方最流行的异步 HTTP 库）
import aiohttp
async def fetch_url(url: str) -> str:
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.text()
async def main():
html = await fetch_url("https://example.com")
print(len(html))httpx（支持同步/异步双模式，API 更友好）
import httpx
async def fetch_url(url: str) -> str:
async with httpx.AsyncClient() as client:
response = await client.get(url)
return response.text
async def main():
html = await fetch_url("https://example.com")
print(len(html))文件 I/O
aiofiles（异步文件操作）
import aiofiles
async def read_write_example():
# 写文件
async with aiofiles.open("example.txt", "w") as f:
await f.write("Hello, 异步文件!\n")
# 读文件
async with aiofiles.open("example.txt", "r") as f:
content = await f.read()
print(content)
