Python的asyncio框架是处理异步IO任务的核心工具，通过事件循环调度协程实现非阻塞执行，而任务生成器的引入可以让批量异步任务的创建和调度更加灵活高效，避免手动逐个创建任务带来的冗余代码和执行效率问题。
asyncio异步执行基础原理
asyncio的核心运行依赖事件循环，事件循环会不断检测可执行的协程任务，当遇到await关键字时，当前协程会暂停执行，让出控制权给事件循环，事件循环转而执行其他就绪的协程，直到等待的资源就绪后再恢复执行。这种机制让单线程内可以并发处理多个IO密集型任务，不会像多线程那样产生线程切换的开销。
普通的异步任务执行通常需要手动创建多个协程并封装成任务，代码示例如下：
import asyncio
async def single_task(task_id):
print(f"任务{task_id}开始执行")
await asyncio.sleep(1) # 模拟IO等待
print(f"任务{task_id}执行完成")
return task_id
async def normal_run():
tasks = []
for i in range(3):
task = asyncio.create_task(single_task(i))
tasks.append(task)
results = await asyncio.gather(*tasks)
print(f"执行结果：{results}")
if __name__ == "__main__":
asyncio.run(normal_run())
任务生成器的工作机制
任务生成器本质上是一个可以动态产出协程对象的可迭代对象，通常基于生成器函数或者异步生成器实现。它的核心作用是把任务的创建逻辑和调度逻辑解耦，不需要提前把所有任务都创建好再提交给事件循环，而是可以在执行过程中按需生成任务，减少内存占用，同时方便对任务进行统一的前置处理或者过滤。
和直接批量创建任务相比，任务生成器的优势在于：
- 支持动态生成任务，不需要提前预知所有任务的数量和参数
- 可以对生成的任务做统一拦截、修改或者优先级排序
- 减少一次性创建大量任务带来的瞬时资源占用
基于任务生成器实现高效异步并发
基础任务生成器实现
首先实现一个简单的同步任务生成器，根据传入的任务参数动态产出协程对象：
def task_generator(task_params):
# 同步生成器，产出协程对象
for param in task_params:
yield single_task(param)
async def run_with_generator():
params = [0, 1, 2, 3]
gen = task_generator(params)
tasks = [asyncio.create_task(coro) for coro in gen]
results = await asyncio.gather(*tasks)
print(f"生成器执行结果：{results}")
异步任务生成器实现
如果需要生成任务的过程中也有异步操作（比如从数据库或者接口获取任务参数），可以使用异步生成器：
import asyncio
async def async_task_generator(task_count):
# 异步生成器，模拟从异步数据源获取任务参数
for i in range(task_count):
await asyncio.sleep(0.1) # 模拟获取参数的异步操作
yield single_task(i)
async def run_with_async_generator():
gen = async_task_generator(4)
tasks = []
async for coro in gen:
tasks.append(asyncio.create_task(coro))
results = await asyncio.gather(*tasks)
print(f"异步生成器执行结果：{results}")
带并发控制的任务生成器
为了避免生成器产出过多任务同时执行导致资源耗尽，可以结合信号量实现并发控制，限制同时运行的任务数量：
import asyncio
async def controlled_task_generator(task_params, max_concurrent=2):
semaphore = asyncio.Semaphore(max_concurrent)
async def wrap_task(param):
async with semaphore:
return await single_task(param)
for param in task_params:
yield wrap_task(param)
async def run_with_controlled_generator():
params = list(range(5))
gen = controlled_task_generator(params, max_concurrent=2)
tasks = [asyncio.create_task(coro) for coro in gen]
results = await asyncio.gather(*tasks)
print(f"带并发控制的执行结果：{results}")
if __name__ == "__main__":
print("===== 普通批量执行 =====")
asyncio.run(normal_run())
print("n===== 同步任务生成器执行 =====")
asyncio.run(run_with_generator())
print("n===== 异步任务生成器执行 =====")
asyncio.run(run_with_async_generator())
print("n===== 带并发控制的任务生成器执行 =====")
asyncio.run(run_with_controlled_generator())
注意事项
使用任务生成器实现异步并发时需要注意几点：
- 生成器产出的协程对象需要及时封装成任务提交给事件循环，避免协程对象未被调度导致无法执行
- 如果任务执行过程中可能抛出异常，使用
asyncio.gather时可以设置return_exceptions=True，避免单个任务异常导致所有任务终止 - 异步生成器需要在异步上下文中迭代，不能使用普通的for循环遍历，必须用
async for语法 - 并发控制的数量需要根据实际业务场景和机器资源调整，过高的并发数可能导致性能下降甚至程序崩溃
适用场景
这种实现方式特别适合以下场景：
- 批量处理大量IO密集型任务，比如批量请求接口、批量读写文件
- 任务参数需要动态获取，无法提前确定所有任务内容
- 需要对任务执行过程做统一管控，比如添加重试逻辑、执行耗时统计