在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也能提供更好的错误处理机制。
注意事项与最佳实践await asyncio.sleep(0)的用途： 它的核心作用是让出当前协程的执行权，允许事件循环调度其他已就绪的协程。它不引入实际的延迟，因此非常适合在紧密的循环中强制事件循环进行上下文切换。
-
任务数量管理： 如果生成器产生任务的速度非常快，而任务本身执行时间较长，可能会导致事件循环中积压大量未完成的任务，消耗过多内存。在这种情况下，可能需要引入一个信号量（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 错误处理： asyncio.TaskGroup在处理组内任务的异常时表现出色。如果组内任何任务抛出异常，TaskGroup会在退出with块时将异常重新抛出，并自动取消组内其他未完成的任务，这简化了复杂的错误管理。
避免过度让步： 虽然await asyncio.sleep(0)很有用，但并非每次创建任务后都需要立即让步。如果你的循环中有其他await操作（例如网络请求、文件I/O），那么这些操作本身就会让出控制权，可能就不需要额外的await asyncio.sleep(0)。判断标准是：如果你的代码段在没有await的情况下运行时间过长，导致其他并发任务得不到执行机会，那就需要考虑让步。
从异步任务生成器中高效地调度任务是asyncio编程中的一个常见需求。核心在于理解asyncio协作式调度的本质，并确保通过await操作主动将控制权让给事件循环。无论是通过简单的await asyncio.sleep(0)，还是利用Python 3.11+提供的asyncio.TaskGroup，都能有效地解决任务创建后无法立即并发执行的问题。选择合适的策略，并结合并发限制和错误处理机制，可以构建出健壮且高性能的异步任务调度系统。
以上就是Python异步任务生成器：实现高效非阻塞任务调度的详细内容，更多请关注知识资源分享宝库其它相关文章！
发表评论:
◎欢迎参与讨论，请在这里发表您的看法、交流您的观点。