使用 asyncio 库在 python 中实现异步后台任务

Question

我希望在后台运行繁重的计算任务而不阻塞 IO。这里的问题是我的主函数不依赖于繁重的任务，并且需要在执行繁重的计算任务之前/同时返回值。举个例子：

def main(args):
   transformed_data_list:List[Dict] = translate_request_to_object(args)
   status = insert_data_into_db(transformed_data:Dict)
   if(status)
      # running background task
      asyncio.run(process_background_task(transformed_data_list))

   # Here, I want to return a success response as soon as data inserted into db
   return "data insert into db"

   async process_background_task(transformed_data_list:List[Dict]):
      for data in transformed_data_list:List:
         asyncio.create_task(heavy_computation_task(data))

但是上面的代码直到

process_background_task

完成才返回响应。

Answer 1

如何在后台运行任务

asyncio.run

是启动事件循环的阻塞函数。如果您想在后台启动

process_background_task

，您需要使用

asyncio.create_task

并使

main

异步。然后运行

asyncio.run(main(...))

。

async def main(args):
    transformed_data_list:List[Dict] = translate_request_to_object(args)
    status = insert_data_into_db(transformed_data:Dict)
    if status:
        # running background task
        asyncio.create_task(process_background_task(transformed_data_list))

    # Here, I want to return a success response as soon as data inserted into db
    return "data insert into db"


async def process_background_task(transformed_data_list:List[Dict]):
    for data in transformed_data_list:List:
        asyncio.create_task(heavy_computation_task(data))

# Start event loop, execute task and wait until task finish.
asyncio.run(main(...))

如何在事件循环中运行繁重的任务

但是在这种情况下

heavy_computation_task

将阻止事件循环，您需要将

ProcessPoolExecutor

与

loop.run_in_executor

一起使用。

这是来自文档的示例：

import asyncio
import concurrent.futures

def cpu_bound():
    # CPU-bound operations will block the event loop:
    # in general it is preferable to run them in a
    # process pool.
    return sum(i * i for i in range(10 ** 7))

async def main():
    loop = asyncio.get_running_loop()

    # Run in a custom process pool:
    with concurrent.futures.ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(
            pool, cpu_bound)
        print('custom process pool', result)

asyncio.run(main())

您的情况：

async def main(args):
    transformed_data_list:List[Dict] = translate_request_to_object(args)
    status = insert_data_into_db(transformed_data:Dict)
    if status:
        # running background task
        asyncio.create_task(process_background_task(transformed_data_list))

    # Here, I want to return a success response as soon as data inserted into db
    return "data insert into db"


async def process_background_task(transformed_data_list:List[Dict]):
    loop = asyncio.get_running_loop()

    for data in transformed_data_list:List:
        with concurrent.futures.ProcessPoolExecutor() as pool:
            await loop.run_in_executor(pool, heavy_computation_task, data)

# Start event loop, execute task and wait until task finish.
asyncio.run(main(...))