如何优雅实现指定数量的异步函数副本并发运行?
优雅实现指定数量异步函数副本并发
可以通过配置驱动的方式动态生成任务列表,不用手动重复编写函数调用,既灵活又易维护,推荐几种实现思路:
方法一:用字典定义函数运行配置
把每个函数的参数和需要运行的副本数存在字典里,遍历字典生成所有任务:
import asyncio async def func_1(a, b): # 模拟无限运行的异步逻辑 while True: print(f"func_1 running with {a}, {b}") await asyncio.sleep(1) async def func_2(b): while True: print(f"func_2 running with {b}") await asyncio.sleep(1) async def func_3(a, c): while True: print(f"func_3 running with {a}, {c}") await asyncio.sleep(1) async def run(): # 定义配置:键是目标函数,值是(参数元组, 副本数量) task_config = { func_1: ((arg1, arg2), 1), func_2: ((arg2,), 2), # 这里指定func_2运行2个副本 func_3: ((arg1, arg3), 1), } # 动态生成任务列表 tasks = [] for func, (args, count) in task_config.items(): tasks.extend([func(*args) for _ in range(count)]) await asyncio.gather(*tasks) if __name__ == "__main__": arg1 = "foo" arg2 = "bar" arg3 = "baz" loop = asyncio.get_event_loop() try: loop.run_until_complete(run()) finally: loop.close()
方法二:用列表配置多参数任务
如果需要给不同副本传递不同参数(比如部分func_2用自定义参数),可以用列表存每个任务组的详细配置:
async def run(): # 列表每个元素是(函数, 参数元组, 副本数) task_groups = [ (func_1, (arg1, arg2), 1), (func_2, (arg2,), 2), # 新增一组不同参数的func_2副本 (func_2, ("custom_param",), 1), (func_3, (arg1, arg3), 1), ] tasks = [] for func, args, count in task_groups: tasks.extend([func(*args) for _ in range(count)]) await asyncio.gather(*tasks)
简化写法:用生成器表达式生成任务
可以用生成器替代循环,让代码更紧凑:
async def run(): task_config = { func_1: ((arg1, arg2), 1), func_2: ((arg2,), 2), func_3: ((arg1, arg3), 1), } # 生成器表达式一次性生成所有任务 tasks = (func(*args) for func, (args, count) in task_config.items() for _ in range(count)) await asyncio.gather(*tasks)
这种方式的核心优势是:用户只需要修改配置里的数字,就能快速调整任意函数的副本数量,完全不用改动asyncio.gather的调用部分,后续新增函数也只需要在配置里加一行即可,扩展性极强。
内容的提问来源于stack exchange,提问作者TimesAndPlaces
相关产品推荐
相关产品推荐

