python - 如何以安全的方式从异步函数调用同步函数

Question

如果一个或多个工作人员同时调用“同步功能”会发生什么？可能有一个或多个 worker 被阻塞了一段时间？

async def worker(queue):
    while True:
        queue_out  = await queue.get()
        file_name = queue_out.file.name
        # Create path + file_name 
        destination_path = create_path(file_name) #<-- SYNC function

        await download_medical(queue_out,destination_path)

async def main():
    queue_in = asyncio.Queue(1)
    workers = [asyncio.create_task(worker(queue_in)) for _ in range(5)]
        
    async for result in get_result(building):
        await queue_in.put(result) 

def create_path(file_name):
    #....#
    #operations related to file and folder on the hdd
    #creates a folder based on file name

Answer 1

简答:

• 如果您从异步协同程序中调用同步(阻塞)函数，则循环中同时运行的所有任务都将停止，直到该函数返回。
• 使用 loop.run_in_executor(...) 在另一个线程或子进程中异步运行阻塞函数。

async def worker(queue):
    loop = Asyncio.get_event_loop()  # get a handle to the current run loop
    while True:
        queue_out  = await queue.get()
        file_name = queue_out.file.name
        
        # run blocking function in an executor
        create_path_task = loop.run_in_executor(None, create_path, file_name)
        destination_path = await create_path_task  # wait for this task to finish

        await download_medical(queue_out, destination_path)

背景:请注意，异步函数(协程)不会并行运行任务，它们会同时运行，这可能看起来是同时运行的。考虑这一点的最简单方法是意识到每次调用 await 时，即在等待结果时，事件循环将暂停当前正在运行的协程并运行另一个协程，直到等待在某事上等等；因此使其协同并发。

等待通常发生在 IO 操作上，因为它们非常耗时且不占用 CPU 资源。 CPU 密集型操作将阻塞循环直到它完成。另请注意，常规 IO 操作本质上是阻塞的，如果您想从并发中受益，则必须使用 Asyncio 兼容库，如 aiofile、aiohttp 等。

关于执行者的更多信息:在不阻塞事件循环的情况下运行常规同步函数的最简单方法是使用 loop.run_in_executor。第一个参数采用 concurrent.futures 模块中的执行器，例如 ThreadPoolExecutor 或 ProcessPoolExecutor。通过传递 None，Asyncio 将自动在默认的 ThreadPoolExecutor 中运行您的函数。如果您的任务是 CPU 密集型任务，请使用 ProcessPoolExecutor 以便它可以使用多个 CPU 核心并真正并行运行。