在Python的异步编程体系中，协程生成器类是一种同时具备协程异步执行能力和生成器迭代输出特性的特殊类，能够优雅地处理异步场景下的流式数据生成需求。我们可以通过特定的语法规则定义这类类，让实例既可以参与asyncio事件循环调度，又可以通过异步迭代的方式逐步产出数据。
协程生成器类的基础定义规则
要定义协程生成器类，核心是实现__aiter__方法和__anext__方法，这两个方法是异步迭代器的核心协议。同时可以在类的方法中使用yield关键字来实现生成器的产出逻辑，结合async关键字声明异步方法。
基础的定义结构如下：
import asyncio
class AsyncGeneratorClass:
def __init__(self, max_count):
# 初始化最大生成数量
self.max_count = max_count
self.current = 0
def __aiter__(self):
# 返回异步迭代器自身
return self
async def __anext__(self):
# 异步下一个元素逻辑
if self.current >= self.max_count:
# 迭代结束抛出异常
raise StopAsyncIteration
# 模拟异步操作，比如等待IO
await asyncio.sleep(0.1)
self.current += 1
# 产出当前值
return self.current
带yield语法的协程生成器类实现
如果需要在类的方法中直接使用yield实现生成器逻辑，同时保留协程特性，可以将__anext__方法定义为异步生成器函数，即在方法内部使用yield关键字，此时方法不需要显式返回迭代值，也不需要手动抛出StopAsyncIteration异常。
import asyncio
class CoroutineGeneratorClass:
def __init__(self, start, end):
self.start = start
self.end = end
def __aiter__(self):
return self
async def __anext__(self):
# 异步生成器函数，使用yield产出值
for i in range(self.start, self.end):
# 模拟异步操作
await asyncio.sleep(0.1)
yield i
# 迭代完成后自动抛出StopAsyncIteration，无需手动处理
return
协程生成器类的使用示例
定义完成协程生成器类之后，我们可以通过async for语法来迭代类的实例，获取生成的数据，同时整个过程是异步执行的，不会阻塞事件循环。
import asyncio
async def main():
# 实例化协程生成器类
generator = CoroutineGeneratorClass(1, 4)
# 异步迭代获取生成的数据
async for value in generator:
print(f"获取到值: {value}")
if __name__ == "__main__":
# 运行异步主函数
asyncio.run(main())
上述代码的输出结果为：
获取到值: 1 获取到值: 2 获取到值: 3
注意事项
- 协程生成器类必须实现
__aiter__方法，该方法需要返回异步迭代器对象，通常返回self即可。 - 如果使用
yield的异步生成器方式实现__anext__方法，不需要手动抛出StopAsyncIteration异常，迭代结束时会自动抛出。 - 协程生成器类的实例只能通过
async for语法进行迭代，不能使用普通的for循环。 - 类中的异步操作需要配合
await关键字使用，确保不会阻塞整个事件循环。
实际应用场景
协程生成器类非常适合处理异步数据流场景，比如从异步数据库查询中逐步获取数据、从异步消息队列中逐步消费消息、异步读取大文件并按行产出内容等。相比普通的同步生成器，它不会在等待IO时阻塞整个程序，能够大幅提升程序的并发处理能力。