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

本文探讨了如何在Python中使用asyncio从生成器高效、非阻塞地调度和执行异步任务。核心在于理解asyncio事件循环的运行机制，通过周期性地将控制权交还给事件循环（例如使用await asyncio.sleep(0)），确保已调度的任务能够获得执行机会。文章还介绍了Python 3.11+中asyncio.TaskGroup的使用，以实现更结构化的并发任务管理。挑战：从生成器调度异步任务的非阻塞执行

在异步编程中，我们经常需要从一个数据源（例如一个生成器）持续获取数据，并为每个数据项启动一个独立的异步任务进行处理。一个常见的误解是，简单地在同步循环中使用asyncio.create_task()就能实现并发执行。然而，create_task()仅仅是将一个协程函数包装成一个任务并将其调度到事件循环中，它本身并不会立即执行该任务，也不会自动将控制权交还给事件循环。

考虑以下场景，我们有一个生成器持续产生数据，并希望为每个数据项启动一个模拟耗时操作的异步任务：

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)
        # 实际应用中可能从外部源（如队列、网络）获取数据

如果manager函数尝试通过同步循环来调度任务，可能会遇到任务无法并发执行的问题：

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，事件循环也无法切换上下文。

在上述manager_initial_attempt中，for循环会快速迭代，连续调用create_task。由于create_task是非阻塞的，manager函数本身并未暂停，因此事件循环没有机会运行这些新创建的任务。这导致任务看起来是同步执行的，或者根本不执行，因为它从未将控制权交还给事件循环。

核心解决方案：显式让出控制权

要解决这个问题，关键在于在每次调度任务后，显式地将控制权交还给asyncio事件循环，哪怕只是短暂的一瞬间。await关键字是实现这一点的核心机制。我们可以使用await asyncio.sleep(0)来达到目的。asyncio.sleep(0)会立即暂停当前协程的执行，并将控制权交还给事件循环。事件循环随后会检查是否有其他已就绪的任务可以运行（包括我们刚刚使用create_task调度的任务），然后再重新安排当前协程的执行。

通过在每次create_task后添加await asyncio.sleep(0)，manager函数就可以确保事件循环有机会执行其他并发任务：

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) # 等待所有任务完成

在这个改进后的manager_with_yield函数中，每次调度一个wrapper任务后，await asyncio.sleep(0)都会让当前manager协程暂停，并允许事件循环运行其他已就绪的任务，包括刚刚创建的wrapper任务。这样就实现了真正的并发调度和执行。

进阶解决方案：使用 asyncio.TaskGroup (Python 3.11+)

对于Python 3.11及更高版本，asyncio.TaskGroup提供了一种更现代、更结构化的方式来管理一组并发任务。它不仅简化了任务的创建和等待，还提供了更好的错误处理机制，确保在任务组退出时所有子任务都已完成或被妥善处理。

使用asyncio.TaskGroup，我们的manager函数可以变得更加简洁和健壮：

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

重要提示： 即使使用asyncio.TaskGroup，在for循环内部从生成器获取数据并创建任务时，仍然需要await asyncio.sleep(0)来周期性地让出控制权。TaskGroup负责任务的生命周期管理和等待，但它并不会改变事件循环的调度机制——即只有当当前运行的协程await时，事件循环才有机会切换到其他任务。

完整示例代码

结合上述概念，以下是一个完整的示例，演示如何从一个生成器非阻塞地调度和执行异步任务，并使用asyncio.TaskGroup进行结构化管理：

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())

运行上述代码，你将看到任务的开始和完成信息是交错出现的，这证明了它们是并发执行的。

注意事项与最佳实践

1. await的重要性： 在asyncio中，只有当一个协程await另一个协程或一个可等待对象时，事件循环才有机会切换到其他已就绪的任务。这是实现并发的关键。
2. asyncio.sleep(0)的用途： 它是将控制权交还给事件循环的最简单方式，即使没有实际的延迟需求，也能确保事件循环有机会处理其他任务。在某些情况下，如果循环体内部已经有其他await操作（例如await queue.get()），那么可能就不需要额外的await asyncio.sleep(0)了。
3. asyncio.TaskGroup：
   • 结构化并发： 强烈推荐在Python 3.11+中使用TaskGroup来管理一组相关的并发任务。它提供了清晰的生命周期管理，并在退出时自动等待所有子任务完成。
   • 错误处理： TaskGroup还能更好地处理子任务中的异常，当一个子任务失败时，它会取消其他任务并重新抛出异常，从而避免资源泄露。
   • 替代方案： 在Python 3.11之前，可以使用asyncio.gather()来等待一组任务，但需要手动收集任务列表。
4. 生成器与异步：
   • 如果你的生成器本身是异步的（例如，它内部有await操作来获取数据），那么它应该是一个异步生成器（async def），并且你需要使用async for来迭代它。在这种情况下，async for循环内部的await操作自然会把控制权交还给事件循环。
   • 如果生成器是同步的（如示例中的generator()），那么在async函数中可以直接使用for ... in ...迭代，但需要手动处理await asyncio.sleep(0)来确保并发。
5. 任务生命周期管理： 对于长时间运行或可能失败的任务，考虑更复杂的任务管理策略，例如设置超时、取消任务或实现重试逻辑。

总结

从生成器非阻塞地调度异步任务是asyncio应用中的一个常见模式。关键在于理解事件循环的工作原理：create_task()仅仅是调度，真正的并发执行需要通过await操作来显式地让出控制权。无论是使用await asyncio.sleep(0)配合asyncio.create_task()，还是采用Python 3.11+中更强大的asyncio.TaskGroup，核心思想都是确保事件循环有机会在任务调度之间切换上下文，从而实现高效的并发处理。掌握这一机制，将能更有效地构建响应迅速、高吞吐量的异步应用。

以上就是Python asyncio：实现从生成器非阻塞地执行异步任务的详细内容，更多请关注知识资源分享宝库其它相关文章！