Python asyncio：从任务生成器实现高效异步并发执行的原理与实践

import asyncio, random 

async def wrapper(word: str):
    print(f"Executing task for: {word}")
    await asyncio.sleep(1) # 模拟耗时操作
    print(f"Finished task for: {word}")

def generator():
    abc = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'
    while True:
        yield random.choice(abc)

async def manager():
    loop = asyncio.get_event_loop()
    for letter in generator():
        loop.create_task(wrapper(letter)) # 创建任务，但不等待
        # 问题在于：这里没有让出控制权，事件循环无法调度其他任务

async def main():
    await manager() # manager是一个无限循环，此处会阻塞

if __name__ == '__main__':
    # asyncio.run(manager()) # 这样调用会因为manager的无限循环而阻塞
    # 需要一种方式让manager能够持续创建任务，同时让其他任务运行
    pass # 暂时不运行，因为会阻塞

import asyncio, random 

async def wrapper(word: str):
    """模拟一个耗时操作的异步任务"""
    print(f"Executing task for: {word}")
    await asyncio.sleep(1) # 任务模拟
    print(f"Finished task for: {word}")

def generator():
    """一个无限生成随机字母的生成器"""
    abc = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'
    while True:
        yield random.choice(abc)

async def manager_with_yield():
    """
    负责从生成器获取任务并调度，通过await asyncio.sleep(0)显式让出控制权。
    """
    loop = asyncio.get_event_loop()
    print("Manager started, generating tasks...")
    for i, letter in enumerate(generator()):
        loop.create_task(wrapper(letter))
        print(f"Task {i+1} created for '{letter}'")
        await asyncio.sleep(0) # 关键：让出控制权，允许其他任务运行
        # 实际应用中，可以根据需要增加一个短暂的等待，例如 await asyncio.sleep(0.01)
        # 或者在处理一定数量任务后才让出控制权，以平衡调度开销和响应性。

        if i >= 10: # 示例：限制生成任务的数量，否则会无限运行
            print("Generated 10 tasks, stopping manager.")
            break

async def main_with_yield():
    """主入口点，运行带有显式让出控制权的manager"""
    await manager_with_yield()
    # 等待所有已创建的wrapper任务完成
    print("Manager finished, waiting for remaining tasks...")
    await asyncio.sleep(2) # 给剩余任务一些时间完成

if __name__ == '__main__':
    print("--- Running Solution 1: Explicit Yielding ---")
    asyncio.run(main_with_yield())
    print("--- Solution 1 Finished ---")

import asyncio, random 

async def wrapper(word: str):
    """模拟一个耗时操作的异步任务"""
    print(f"Executing task for: {word}")
    await asyncio.sleep(1) # 任务模拟
    print(f"Finished task for: {word}")

def generator():
    """一个无限生成随机字母的生成器"""
    abc = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'
    while True:
        yield random.choice(abc)

async def manager_with_taskgroup():
    """
    负责从生成器获取任务并调度，使用asyncio.TaskGroup进行结构化并发管理。
    """
    print("Manager started with TaskGroup, generating tasks...")
    async with asyncio.TaskGroup() as tg: # 使用TaskGroup上下文管理器
        for i, letter in enumerate(generator()):
            tg.create_task(wrapper(letter)) # 在TaskGroup中创建任务
            print(f"Task {i+1} created for '{letter}'")
            # TaskGroup在内部会处理调度和让出控制权，通常无需额外的await asyncio.sleep(0)
            # 但如果生成任务的速度极快，且任务本身耗时很短，
            # 偶尔添加 await asyncio.sleep(0) 仍可能优化响应性。

            if i >= 10: # 示例：限制生成任务的数量
                print("Generated 10 tasks, stopping manager.")
                break
    print("TaskGroup exited. All tasks created within it should have completed or been cancelled.")

async def main_with_taskgroup():
    """主入口点，运行带有TaskGroup的manager"""
    await manager_with_taskgroup()

if __name__ == '__main__':
    print("\n--- Running Solution 2: Using asyncio.TaskGroup (Python 3.11+) ---")
    # 确保Python版本 >= 3.11
    if hasattr(asyncio, 'TaskGroup'):
        asyncio.run(main_with_taskgroup())
    else:
        print("Warning: asyncio.TaskGroup requires Python 3.11 or later. Skipping this example.")
    print("--- Solution 2 Finished ---")

import asyncio
import random
import time

async def process_item(item_id: int, data: str):
    """模拟一个异步处理任务，打印处理信息并模拟耗时"""
    start_time = time.time()
    print(f"[{item_id}] Processing item: '{data}'...")
    await asyncio.sleep(random.uniform(0.5, 2.0)) # 模拟随机耗时
    end_time = time.time()
    print(f"[{item_id}] Finished item: '{data}' in {end_time - start_time:.2f}s")

def item_generator(max_items: int = 20):
    """一个生成器，生成带ID的随机数据"""
    abc = 'abcdefghijklmnopqrstuvwxyz'
    for i in range(1, max_items + 1):
        yield i, random.choice(abc) * random.randint(3, 8) # 生成随机长度的字符串

async def task_dispatcher():
    """
    任务调度器，从生成器获取数据并创建异步任务。
    优先使用TaskGroup，如果不可用则回退到显式让出控制权。
    """
    print("--- Task Dispatcher Started ---")
    item_count = 0

    if hasattr(asyncio, 'TaskGroup'):
        print("Using asyncio.TaskGroup for task management.")
        async with asyncio.TaskGroup() as tg:
            for item_id, data in item_generator():
                tg.create_task(process_item(item_id, data))
                print(f"Dispatched task {item_id} for data '{data}'")
                item_count += 1
                # 即使使用TaskGroup，如果生成任务的速度远超任务执行速度，
                # 也可以考虑在此处加入一个短暂的await，以避免内存中积压过多未开始的任务。
                # 例如： if item_count % 5 == 0: await asyncio.sleep(0.01)
        print(f"--- TaskGroup Finished. All {item_count} tasks completed or cancelled. ---")
    else:
        print("asyncio.TaskGroup not available (Python < 3.11). Falling back to explicit yield.")
        loop = asyncio.get_event_loop()
        for item_id, data in item_generator():
            loop.create_task(process_item(item_id, data))
            print(f"Dispatched task {item_id} for data '{data}'")
            item_count += 1
            await asyncio.sleep(0) # 显式让出控制权
        print(f"--- Dispatcher Finished creating {item_count} tasks. Waiting for them to complete. ---")
        # 由于是手动创建任务且没有TaskGroup等待，需要额外等待所有任务完成
        await asyncio.sleep(3) # 粗略等待，实际应用中可能需要更精细的等待机制

async def main():
    """主程序入口"""
    await task_dispatcher()
    print("All dispatching and processing should be complete.")

if __name__ == '__main__':
    asyncio.run(main())