如何编写两个并发异步并行函数并与另一个执行一个？答案

【问题标题】：How to write two concurrent async parallel function and execute one with another?如何编写两个并发异步并行函数并与另一个执行一个？
【发布时间】：2021-05-10 21:48:32
【问题描述】：

如何拥有两个并发的进程池，来加速功能，

我正在尝试从 func 函数内部并行执行 test 函数，

有没有更好的方法可以在另一个并行并发异步函数中运行？

注意：这是我原始函数的一个基本示例，我正在寻找并行化另一个并行函数的方法。

import concurrent.futures as cf
import multiprocessing as mp
import asyncio


def arg(x):
    print("Hello", x)


async def test(x):
    print(x)
    asyncio_loop = asyncio.get_event_loop()
    with cf.ProcessPoolExecutor(6) as pool:
        results = [asyncio_loop.run_in_executor(pool, arg, x) for x in range(10)]
        results = await asyncio.gather(*results)
        return results


async def func():
    asyncio_loop = asyncio.get_event_loop()
    with cf.ProcessPoolExecutor(6) as loop:
        results = [asyncio_loop.run_in_executor(loop, test, x) for x in range(10)]
        results = await asyncio.gather(*results)
        return results


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(func())
    loop.close()

这是错误，

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "testing.py", line 29, in <module>
    loop.run_until_complete(func())
  File "/usr/lib/python3.8/asyncio/base_events.py", line 616, in run_until_complete
    return future.result()
  File "testing.py", line 23, in func
    results = await asyncio.gather(*results)
TypeError: cannot pickle 'coroutine' object

【问题讨论】：

标签： python python-3.x python-asyncio python-multiprocessing concurrent.futures

【解决方案1】：

您不需要执行额外的步骤来并行化一个已经异步的函数。由于test 是异步的，因此编写func 的正确方法是：

async def func():
    return await asyncio.gather(*[test(x) for x in range(10)])

另外，请注意，每次调用函数时创建一个进程池意味着进程将重新启动。将池设为全局并在每次需要时重用它（没有with 语句）是一个更好的主意。这样，已经启动的进程将被重用，并形成类名所指的“池”。

【讨论】：