python - 如何并发多次运行相同的异步函数？

Question

设置:

所以我有一个由客户端对象组成的客户端数组。还有两个异步函数，一个用于处理来自 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 数组 中第二个客户端对象 的function_b 进程在收到 WebSocket 消息后大约需要 1000 毫秒才能完成。显然，随着每次调用，这种延迟会变大。

解决方案？

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

Answer 1

要从协同例程中并行启动代码，您必须为其创建一个“任务”(并暂停，不必等待任务)。

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

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 绑定(bind)代码，只能通过在不同线程或进程中运行函数来实现并行。