挑战：从生成器调度异步任务的非阻塞执行
在异步编程中，我们经常需要从一个数据源（例如一个生成器）持续获取数据，并为每个数据项启动一个独立的异步任务进行处理。一个常见的误解是，简单地在同步循环中使用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调度的任务），然后再重新安排当前协程的执行。
立即学习“Python免费学习笔记（深入）”；
通过在每次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())运行上述代码，你将看到任务的开始和完成信息是交错出现的，这证明了它们是并发执行的。
注意事项与最佳实践
- await的重要性： 在asyncio中，只有当一个协程await另一个协程或一个可等待对象时，事件循环才有机会切换到其他已就绪的任务。这是实现并发的关键。
- asyncio.sleep(0)的用途： 它是将控制权交还给事件循环的最简单方式，即使没有实际的延迟需求，也能确保事件循环有机会处理其他任务。在某些情况下，如果循环体内部已经有其他await操作（例如await queue.get()），那么可能就不需要额外的await asyncio.sleep(0)了。
-
asyncio.TaskGroup：
- 结构化并发： 强烈推荐在Python 3.11+中使用TaskGroup来管理一组相关的并发任务。它提供了清晰的生命周期管理，并在退出时自动等待所有子任务完成。
- 错误处理： TaskGroup还能更好地处理子任务中的异常，当一个子任务失败时，它会取消其他任务并重新抛出异常，从而避免资源泄露。
- 替代方案： 在Python 3.11之前，可以使用asyncio.gather()来等待一组任务，但需要手动收集任务列表。
-
生成器与异步：
- 如果你的生成器本身是异步的（例如，它内部有await操作来获取数据），那么它应该是一个异步生成器（async def），并且你需要使用async for来迭代它。在这种情况下，async for循环内部的await操作自然会把控制权交还给事件循环。
- 如果生成器是同步的（如示例中的generator()），那么在async函数中可以直接使用for ... in ...迭代，但需要手动处理await asyncio.sleep(0)来确保并发。
- 任务生命周期管理： 对于长时间运行或可能失败的任务，考虑更复杂的任务管理策略，例如设置超时、取消任务或实现重试逻辑。
总结
从生成器非阻塞地调度异步任务是asyncio应用中的一个常见模式。关键在于理解事件循环的工作原理：create_task()仅仅是调度，真正的并发执行需要通过await操作来显式地让出控制权。无论是使用await asyncio.sleep(0)配合asyncio.create_task()，还是采用Python 3.11+中更强大的asyncio.TaskGroup，核心思想都是确保事件循环有机会在任务调度之间切换上下文，从而实现高效的并发处理。掌握这一机制，将能更有效地构建响应迅速、高吞吐量的异步应用。
