如何同时运行相同的异步函数多次？

Question

设置：

所以我有一个 clients 数组，其中包含 client 对象。还有两个异步函数，一个是处理来自 WebSocket 的传入消息，另一个是使用给定的客户端对象执行一些 API 调用。

当某个消息从 WebSocket 连接到达时，function_a 应该检测到它并在 clients 数组 中启动一个 for 循环。对于 clients 数组 中的每个 client 对象，我想触发 function_b。但是 function_b 应该为每个 client 并发运行。

示例代码：

clients = {}

async def function_a(ws_message): # this is being called each time when a new message arrives
    if ws_message == 'signal_message':
        for client in clients:
             await function_b(client)

async def function_b(client):
    # get client parameters
    # make an API request with those parameters

更新 || asyncio.gather()

的代码

所以我已经用 asyncio.gather() 更新了我的实际代码，但每个函数任务之间仍有约 500 毫秒的延迟。我不确定，但我想这是因为我的 function_b 有一些非异步阻塞代码，这要归功于 requests 库。

clients = {}
tasks = []

async def function_a(ws_message): 
    if ws_message == 'signal_message':
        for client in clients:
            task = asyncio.ensure_future(function_b(client))
            tasks.append(task)
    if tasks:
        await asyncio.gather(*tasks, return_exceptions=True)

async def function_b(client):
    # get client parameters
    # make an API request with those parameters

正如我所说，这也不起作用。

问题：

现在 function_b 被触发没有任何问题，但是 function_b 运行大约 500 毫秒，并且来自 function_a 的 await 调用等待那个在它完成之前。例如，clients 数组 中第二个 client 对象 的 function_b 进程在收到 WebSocket 消息后需要大约 1000 毫秒才能完成。而且这种延迟显然会随着每次调用而变大。

解决方案？

是否可以解决此问题，以便每个 function_b 进程可以同时运行？提前致谢。

Answer 1

要从协同程序中并行启动代码，您必须使用它创建一个“任务”（并暂停，不必等待任务）。

asyncio.gather 调用执行包装、调用、执行和包装 一次性返回值：

clients = {}

async def function_a(ws_message): # this is being called each time when a new message arrives
    if ws_message == 'signal_message':
        await asyncio.gather(*(function_b(client) for client in clients))

交互式提示的完整示例：

In [8]: import asyncio

In [9]: async def coro1():
   ...:     await asyncio.sleep(1)
   ...:     print("step 1")
   ...:     await asyncio.sleep(1)
   ...:     print("step 2")
   ...: 
   ...: 
   ...: 

In [10]: async def coro2():
    ...:     await asyncio.gather(*(coro1() for i in range(5)))
    ...: 

In [11]: asyncio.run(coro2())
step 1
step 1
step 1
step 1
step 1
step 2
step 2
step 2
step 2
step 2

可以看出，所有协程并行运行，输出第一次打印的结果，1s后，显示第二次打印调用的输出。

如果没有观察到并行性，可能是由于协程您没有将执行交给主循环本身等待其他协程。对于 CPU 密集型代码，只能通过在不同的线程或进程中运行函数来实现并行。