如何让两个异步方法同时运行？Python asyncio 实践问题

本人惭愧地承认，使用Python的asyncio已有很长时间，但并未真正理解其工作原理，如今陷入了困境。我的当前程序伪代码如下：

async def api_function1(parameters):
    result = await asyncio.gather(*[some_other_thing(p) for p in parameters])

async def api_function2(parameters):
    result = await asyncio.gather(*[some_other_thing2(p) for p in parameters])    

def a(initial_parameters): 
    output = []
    data = asyncio.run(api_function1(initial_parameters))
    output.append(data)
    while True: 
        data = asyncio.run(api_function1(get_parameters_from_data(data)))
        output.append(data)
        if some_condition is True:
            break
    return output 

def b(initial_parameters): 
    output = []
    data = asyncio.run(api_function2(initial_parameters))
    output.append(data)
    while True: 
        data = asyncio.run(api_function2(get_parameters_from_data(data)))
        output.append(data)
        if some_condition is True:
            break
    return output     

其中a()和b()分别从不同REST API端点获取数据，各自有速率限制和特性。我希望让a()和b()同时运行。

我曾尝试将a()和b()都改为异步方法，并尝试同时await它们，示例如下：

async def a(initial_parameters):
    ...

async def b(initial_parameters):
    ...

A = await a(initial_parameters)
B = await b(initial_parameters)

但这种方式并未奏效。根据文档推测，可能需要手动获取事件循环并将其作为参数传递给a()和b()，再传递给api_function1()和api_function2()，并在任务完成后手动关闭，但不确定是否正确及具体实现方式。同时也欢迎提供更优的设计模式建议。

核心问题分析 #

你之前的尝试无效的原因是：直接await a()后再await b()是串行执行，程序会先等待a()完全结束才会开始b()，并没有实现并行。而原同步版本的a()和b()内部反复调用asyncio.run()，每次都会创建新的事件循环，完全割裂了两个任务的并行可能。

最简调整方案 #

只需要两步就能实现并行：

1. 将a()和b()改为真正的异步函数，移除内部的asyncio.run()，直接await异步API函数：

   async def a(initial_parameters): 
       output = []
       data = await api_function1(initial_parameters)
       output.append(data)
       while True: 
           data = await api_function1(get_parameters_from_data(data))
           output.append(data)
           if some_condition is True:
               break
       return output 
   
   async def b(initial_parameters): 
       output = []
       data = await api_function2(initial_parameters)
       output.append(data)
       while True: 
           data = await api_function2(get_parameters_from_data(data))
           output.append(data)
           if some_condition is True:
               break
       return output     
   

2. 用asyncio.gather()或create_task()并行运行两个异步任务：
   编写一个入口异步函数，把a()和b()包装成任务同时执行：

   async def main():
       # 创建两个并行任务
       task_a = asyncio.create_task(a(your_initial_params_a))
       task_b = asyncio.create_task(b(your_initial_params_b))
       
       # 等待两个任务都完成，获取结果
       result_a, result_b = await asyncio.gather(task_a, task_b)
       
       # 后续处理结果
       print("任务a结果:", result_a)
       print("任务b结果:", result_b)
   
   if __name__ == "__main__":
       # 启动主事件循环
       asyncio.run(main())
   

关键说明 #

• 不需要手动传递或管理事件循环，asyncio.run()会自动创建并管理主循环，异步函数内部会自动绑定当前循环。
• asyncio.create_task()会立即将任务加入事件循环调度，两个任务会在同一个循环里交替执行（实现并行效果），直到都完成。
• 如果其中一个任务抛出异常，gather()会立即传播异常，若需要忽略或单独处理异常，可以给gather()加return_exceptions=True参数。

额外建议（针对速率限制） #

因为两个任务并行执行，需要确保各自的API调用符合速率限制：

• 在some_other_thing()和some_other_thing2()内部添加延迟逻辑（比如await asyncio.sleep(1)），控制请求频率。
• 可以使用异步速率限制库（如aiometer），或者自己实现基于令牌桶的限速器，确保每个任务的请求速率不超过API要求。

内容的提问来源于stack exchange，提问作者cmaz