Python异步任务生成器怎么用

异步任务生成器的挑战

在Python的asyncio框架中，我们经常需要从一个源（例如数据流、事件队列或生成器）持续地创建并启动异步任务。一个常见的场景是，我们有一个任务生成器不断地产生新的工作项，而我们希望将这些工作项封装成异步任务并添加到事件循环中，但又不想立即等待每个任务完成。例如，在一个长轮询（long-polling）服务器中，我们可能希望异步处理接收到的用户事件，而不阻塞后续事件的接收。

考虑以下代码示例，它尝试从一个简单的字符生成器中创建并启动异步任务：

import asyncio, random

async def wrapper(word: str):
    """模拟一个异步任务，打印字符并暂停1秒。"""
    print(f"处理任务: {word}")
    await asyncio.sleep(1) # 任务模拟，实际可能涉及I/O操作

def generator():
    """一个无限生成器，每次产生一个随机字符。"""
    abc = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'
    while True:
        yield random.choice(abc)

async def manager_sync_issue():
    """尝试异步创建任务，但存在同步执行问题。"""
    loop = asyncio.get_event_loop()
    print("开始任务生成...")
    for letter in generator():
        print(f"生成任务: {letter}")
        loop.create_task(wrapper(letter)) # 这里创建了任务
        # 问题：当前循环没有让出控制权，导致wrapper任务无法运行
        if random.random() < 0.05: # 随机停止，避免无限循环
            break
    print("任务生成结束。")

# asyncio.run(manager_sync_issue()) # 运行会发现，wrapper任务不会真正并行执行

尽管我们使用了loop.create_task(wrapper(letter))来创建异步任务，但实际运行上述manager_sync_issue函数会发现，wrapper任务并不会真正并行执行。输出会持续打印“生成任务: X”，直到生成器停止，然后才开始打印“处理任务: X”。这是因为manager_sync_issue中的for循环是一个同步操作，它不断地创建任务，但从未主动让出控制权给事件循环，导致事件循环没有机会调度和运行那些已经创建的wrapper任务。

核心原理：主动让出控制权

asyncio中的并发与传统的多线程编程有本质区别。在多线程中，操作系统负责线程的调度和上下文切换。而在asyncio中，协程的调度是协作式的，这意味着一个协程必须主动地通过await操作将控制权让给事件循环，事件循环才能有机会调度其他准备就绪的协程。

即使我们通过create_task创建了一个新的协程任务，如果当前运行的协程（例如manager_sync_issue）没有执行任何await操作来让出控制权，事件循环就无法介入并执行新创建的任务。await asyncio.sleep(0)是一个非常巧妙且常用的技巧，它不会引入任何实际的延迟，但会强制当前协程将控制权让给事件循环，允许事件循环去调度和运行其他已准备好的任务。

解决方案一：手动让出控制权

最直接的解决方案是在创建任务后，立即通过await asyncio.sleep(0)让出控制权。这使得事件循环有机会在下一次迭代中运行已创建的wrapper任务，从而实现真正的并发。

import asyncio, random

async def wrapper(word: str):
    """模拟一个异步任务，打印字符并暂停1秒。"""
    print(f"处理任务: {word}")
    await asyncio.sleep(1) # 任务模拟

def generator():
    """一个无限生成器，每次产生一个随机字符。"""
    abc = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'
    while True:
        yield random.choice(abc)

async def manager_manual_yield():
    """通过await asyncio.sleep(0)实现异步任务的并发调度。"""
    loop = asyncio.get_event_loop()
    print("开始任务生成...")
    tasks = [] # 可以选择保存任务引用，但对于本例并非必须
    for i, letter in enumerate(generator()):
        print(f"生成任务: {letter}")
        task = loop.create_task(wrapper(letter))
        tasks.append(task)
        await asyncio.sleep(0) # 关键：让出控制权，允许其他任务运行
        if i >= 10: # 限制生成任务数量，避免无限运行
            break
    print("任务生成结束，等待所有任务完成...")
    await asyncio.gather(*tasks) # 等待所有生成的任务完成
    print("所有任务完成。")

# asyncio.run(manager_manual_yield())

运行manager_manual_yield后，你会看到“生成任务”和“处理任务”的输出交替出现，这表明wrapper任务正在与manager_manual_yield协程并发执行。

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

对于Python 3.11及更高版本，asyncio.TaskGroup提供了一种更结构化、更健壮的方式来管理一组相关的异步任务。它是一个上下文管理器，可以确保在退出TaskGroup块时，所有在其内部创建的任务都已完成或被取消。结合await asyncio.sleep(0)，TaskGroup能更好地组织和控制任务。

import asyncio, random

async def wrapper(word: str):
    """模拟一个异步任务，打印字符并暂停1秒。"""
    print(f"处理任务: {word}")
    await asyncio.sleep(1) # 任务模拟

def generator():
    """一个无限生成器，每次产生一个随机字符。"""
    abc = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'
    while True:
        yield random.choice(abc)

async def manager_with_taskgroup():
    """使用asyncio.TaskGroup和await asyncio.sleep(0)实现并发调度。"""
    print("开始任务生成...")
    # TaskGroup是一个上下文管理器，自动管理其内部创建的任务
    async with asyncio.TaskGroup() as tg:
        for i, letter in enumerate(generator()):
            print(f"生成任务: {letter}")
            tg.create_task(wrapper(letter)) # 使用TaskGroup创建任务
            await asyncio.sleep(0) # 关键：让出控制权
            if i >= 10: # 限制生成任务数量
                break
    print("任务生成结束，TaskGroup已等待所有任务完成。")

# 完整运行示例
if __name__ == "__main__":
    print("\n--- 运行 manager_with_taskgroup ---")
    asyncio.run(manager_with_taskgroup())

在这个manager_with_taskgroup示例中，async with asyncio.TaskGroup() as tg: 确保了在退出with块时，tg内所有由tg.create_task()启动的任务都会被等待完成。这使得任务管理更加清晰，并且在任务出现异常时，TaskGroup也能提供更好的错误处理机制。

注意事项与最佳实践

1. await asyncio.sleep(0)的用途： 它的核心作用是让出当前协程的执行权，允许事件循环调度其他已就绪的协程。它不引入实际的延迟，因此非常适合在紧密的循环中强制事件循环进行上下文切换。

2. 任务数量管理： 如果生成器产生任务的速度非常快，而任务本身执行时间较长，可能会导致事件循环中积压大量未完成的任务，消耗过多内存。在这种情况下，可能需要引入一个信号量（Semaphore）或有界队列（Bounded Queue）来限制并发任务的数量，例如：

   # 示例：使用信号量限制并发
   async def manager_with_semaphore():
       sem = asyncio.Semaphore(5) # 限制最多5个wrapper任务同时运行
       async def bounded_wrapper(word: str):
           async with sem: # 获取信号量，限制并发
               await wrapper(word)
   
       async with asyncio.TaskGroup() as tg:
           for i, letter in enumerate(generator()):
               tg.create_task(bounded_wrapper(letter))
               await asyncio.sleep(0)
               if i >= 20:
                   break

3. 错误处理： asyncio.TaskGroup在处理组内任务的异常时表现出色。如果组内任何任务抛出异常，TaskGroup会在退出with块时将异常重新抛出，并自动取消组内其他未完成的任务，这简化了复杂的错误管理。

4. 避免过度让步： 虽然await asyncio.sleep(0)很有用，但并非每次创建任务后都需要立即让步。如果你的循环中有其他await操作（例如网络请求、文件I/O），那么这些操作本身就会让出控制权，可能就不需要额外的await asyncio.sleep(0)。判断标准是：如果你的代码段在没有await的情况下运行时间过长，导致其他并发任务得不到执行机会，那就需要考虑让步。

总结

从异步任务生成器中高效地调度任务是asyncio编程中的一个常见需求。核心在于理解asyncio协作式调度的本质，并确保通过await操作主动将控制权让给事件循环。无论是通过简单的await asyncio.sleep(0)，还是利用Python 3.11+提供的asyncio.TaskGroup，都能有效地解决任务创建后无法立即并发执行的问题。选择合适的策略，并结合并发限制和错误处理机制，可以构建出健壮且高性能的异步任务调度系统。