Python异步生成器好用吗

wen python案例 2026-07-12 1

本文目录导读：

Python异步生成器好用吗

核心优势
使用注意事项
何时应该使用？

Python 的异步生成器（async for + yield）非常好用，但在正确的场景下才能发挥最大价值，它解决了传统同步生成器在 I/O 密集型任务中的阻塞问题，是处理流式数据、长轮询、分页 API、实时数据管道等场景的利器。

核心优势

非阻塞的延迟生产数据

传统生成器遇到 time.sleep() 或 requests.get() 会阻塞整个事件循环，异步生成器可以优雅地 await 其他协程。

示例：模拟实时流式数据

import asyncio
async def sensor_data_stream():
    """模拟传感器每秒产生一个数据点"""
    for i in range(10):
        await asyncio.sleep(1)  # 非阻塞等待
        yield {"sensor_id": 1, "value": i * 0.5, "timestamp": asyncio.get_event_loop().time()}
async def consumer():
    async for data in sensor_data_stream():
        print(f"收到数据: {data}")
        # 这里也可以 await 其他操作，比如写入数据库
asyncio.run(consumer())

关键点：await asyncio.sleep(1) 不会阻塞其他协程,事件循环可以处理其他任务。

处理分页 API 或流式响应

当你需要从远程 API 拉取大量分页数据时,异步生成器非常优雅。

示例：分页读取 GitHub Star 用户列表（简化版）

import aiohttp
import asyncio
async def get_stargazers(owner: str, repo: str):
    async with aiohttp.ClientSession() as session:
        url = f"https://api.github.com/repos/{owner}/{repo}/stargazers"
        page = 1
        while True:
            params = {"page": page, "per_page": 100}
            async with session.get(url, params=params) as resp:
                if resp.status != 200:
                    break
                data = await resp.json()
                if not data:
                    break
                for user in data:
                    yield user["login"]
                page += 1
                # 避免被限速
                await asyncio.sleep(0.5)
async def main():
    async for username in get_stargazers("python", "cpython"):
        print(f"Star用户: {username}")
        # 可以限制只打印前10个
        # if username_counter >= 10: break
asyncio.run(main())

优势：边拉取边消费，无需将所有用户先加载到内存中,内存友好。

实现 WebSocket 消息流

import asyncio
import websockets
async def listen_to_websocket(url: str):
    async with websockets.connect(url) as ws:
        async for message in ws:
            yield message  # 每次有新消息就产生
async def process_messages():
    async for msg in listen_to_websocket("wss://example.com/feed"):
        processed = await process(msg)  # 可以异步处理
        await save_to_db(processed)  # 可以异步写库

与同步对比：同步版本需要手动循环 recv(),且无法在等待消息时执行其他任务。

使用注意事项

只能在异步函数中使用 `yield`

async def good():
    for i in range(10):
        await asyncio.sleep(0.1)
        yield i  # ✅ 正确
def bad():
    for i in range(10):
        yield i  # 这是同步生成器，不是异步生成器

错误示例：在同步函数中使用 async yield 会直接报语法错误。

消费者必须使用 `async for`

async_gen = my_async_generator()
# ❌ 错误：for item in async_gen:  # TypeError: 'async_generator' object is not iterable
# ✅ 正确
async for item in async_gen:
    ...

如果需要在同步代码中消费，可以使用 asyncio.run() + async list comprehension：

result = asyncio.run([item async for item in my_async_generator()])

异常处理和清理

异步生成器支持 try/finally，但清理工作需要手动触发（通过 aclose()）或依赖 async with：

async def resource_stream():
    try:
        resource = await acquire_resource()
        while True:
            data = await resource.read()
            if not data:
                break
            yield data
    finally:
        await resource.close()  # 确保清理
# 使用 `async with` 自动管理
async def consumer():
    async with resource_stream() as stream:
        async for chunk in stream:
            process(chunk)

没有 async with 时，需要手动 await agen.aclose()，否则 finally 可能不会执行。