Python asyncio：实现从生成器非阻塞地执行异步任务

import asyncio
import random

async def wrapper(word: str):
    """模拟一个耗时的异步任务"""
    print(f"开始处理: {word}")
    await asyncio.sleep(random.uniform(0.5, 2)) # 模拟异步I/O或计算
    print(f"完成处理: {word}")

def generator():
    """一个无限生成器，持续产生随机字母"""
    abc = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'
    while True:
        yield random.choice(abc)
        # 实际应用中可能从外部源（如队列、网络）获取数据

async def manager_initial_attempt():
    loop = asyncio.get_event_loop()
    print("开始调度任务 (初始尝试)")
    for _ in range(10): # 仅为示例，实际生成器是无限的
        letter = next(generator())
        loop.create_task(wrapper(letter)) # 任务被调度，但事件循环未获得执行机会
    print("所有任务已调度 (初始尝试)")
    # 此时，如果manager_initial_attempt不await任何东西，
    # 整个程序可能会在任务完成前退出，或者任务不会并发执行。
    # 即使程序不退出，由于没有await，事件循环也无法切换上下文。

async def manager_with_yield():
    loop = asyncio.get_event_loop()
    print("开始调度任务 (显式让出控制权)")
    tasks = []
    for _ in range(10): # 仅为示例，实际生成器是无限的
        letter = next(generator())
        task = loop.create_task(wrapper(letter))
        tasks.append(task)
        await asyncio.sleep(0) # 关键：让出控制权给事件循环
    print("所有任务已调度 (显式让出控制权)")
    await asyncio.gather(*tasks) # 等待所有任务完成

async def manager_with_taskgroup():
    print("开始调度任务 (使用 TaskGroup)")
    # TaskGroup 提供了一个上下文管理器，在其退出时会自动等待所有子任务完成
    async with asyncio.TaskGroup() as tg:
        for _ in range(10): # 仅为示例，实际生成器是无限的
            letter = next(generator())
            tg.create_task(wrapper(letter)) # 使用 TaskGroup 创建任务
            await asyncio.sleep(0) # 仍然需要让出控制权
    print("所有任务已调度并完成 (使用 TaskGroup)")
    # TaskGroup 退出时会自动等待所有任务完成，无需额外的 asyncio.gather

import asyncio
import random

async def wrapper(word: str):
    """
    模拟一个耗时的异步任务。
    打印开始和完成信息，并模拟随机延迟。
    """
    delay = random.uniform(0.5, 2)
    print(f"[{asyncio.current_task().get_name()}] 开始处理: {word} (预计 {delay:.2f}s)")
    await asyncio.sleep(delay)
    print(f"[{asyncio.current_task().get_name()}] 完成处理: {word}")

def generator():
    """
    一个无限生成器，持续产生随机字母。
    在实际应用中，这可能是一个从外部源（如消息队列、API）获取数据的协程。
    """
    abc = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'
    while True:
        yield random.choice(abc)
        # 在实际的异步生成器中，这里可能有一个 await 操作来获取下一个数据
        # 例如: await queue.get()

async def manager():
    """
    主管理器协程，负责从生成器获取数据并调度异步任务。
    使用 asyncio.TaskGroup 实现结构化并发。
    """
    print("--- 启动任务管理器 ---")
    # 使用 TaskGroup 确保所有子任务在管理器退出前完成
    async with asyncio.TaskGroup() as tg:
        print("TaskGroup 已启动，开始调度任务...")
        # 循环从生成器获取数据并创建任务
        for i in range(15): # 模拟处理15个事件
            letter = next(generator()) # 从同步生成器获取数据
            # 创建一个异步任务并添加到 TaskGroup
            # 为任务指定名称，方便调试
            tg.create_task(wrapper(letter), name=f"Task-{i:02d}-{letter}")

            # 关键步骤：让出控制权给事件循环
            # 允许其他已调度的任务（包括刚刚创建的 wrapper 任务）运行
            await asyncio.sleep(0) 

            # 如果需要，可以在这里添加一些日志或条件判断
            # print(f"已调度任务 {i+1} for letter '{letter}'")

    print("--- 所有任务已调度并完成 ---")

if __name__ == "__main__":
    # 运行主管理器协程
    # asyncio.run() 会启动事件循环并运行顶层协程
    asyncio.run(manager())