FastAPI场景下如何将同步第三方库API转为异步并行调用

我目前用同步第三方库以同步方式调用4个API请求，想通过并行执行缩短总耗时。用FastAPI做微框架，现有代码如下：

utilities.py #

async def get_api_1_data():
    data = some_third_party_lib()
    return data

async def get_api_2_data():
    data = some_third_party_lib()
    return data

async def get_api_3_data():
    data = some_third_party_lib()
    return data

async def get_api_4_data():
    data = some_third_party_lib()
    return data

main.py #

import asyncio
from fastapi import FastAPI
app = FastAPI()

@app.get("/")
async def fetch_new_exposure_api_data(node: str):
   functions_to_run = [get_api_1_data(), get_api_2_data(), get_api_3_data(), get_api_4_data()]
   r1, r2, r3, r4 = await asyncio.gather(*functions_to_run)
   return [r1, r2, r3, r4]

问题是没法给同步的some_third_party_lib()加await，有没有办法把它转成异步实现并行调用？

因为some_third_party_lib()是同步阻塞函数，直接放在async函数里会卡死整个事件循环，导致asyncio.gather根本没法真正并行。可以用下面两种方式解决：

方法一：用asyncio.run_in_executor把同步函数扔到线程池执行 #

修改utilities.py里的函数，把同步调用委托给线程池，这样事件循环能在等同步函数执行的时候处理其他任务：

import asyncio
from concurrent.futures import ThreadPoolExecutor

# 全局定义线程池，避免每次创建开销
executor = ThreadPoolExecutor(max_workers=4)

# 先把原来的异步函数改成纯同步的
def get_api_1_data_sync():
    return some_third_party_lib()

def get_api_2_data_sync():
    return some_third_party_lib()

def get_api_3_data_sync():
    return some_third_party_lib()

def get_api_4_data_sync():
    return some_third_party_lib()

# 再包装成可await的异步函数
async def get_api_1_data():
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(executor, get_api_1_data_sync)

async def get_api_2_data():
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(executor, get_api_2_data_sync)

async def get_api_3_data():
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(executor, get_api_3_data_sync)

async def get_api_4_data():
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(executor, get_api_4_data_sync)

main.py不用改，继续用asyncio.gather并行调用这些异步函数就行，这时候四个同步请求会在不同线程里跑，总耗时差不多等于单个请求的耗时。

方法二：直接在main.py里包装同步函数 #

要是不想改utilities.py，也能在main.py里直接把同步函数包装成可await的对象：

import asyncio
from concurrent.futures import ThreadPoolExecutor
from fastapi import FastAPI
from utilities import get_api_1_data, get_api_2_data, get_api_3_data, get_api_4_data

app = FastAPI()
executor = ThreadPoolExecutor(max_workers=4)

@app.get("/")
async def fetch_new_exposure_api_data(node: str):
    loop = asyncio.get_running_loop()
    # 注意这里传函数本身，别加括号调用
    tasks = [
        loop.run_in_executor(executor, get_api_1_data),
        loop.run_in_executor(executor, get_api_2_data),
        loop.run_in_executor(executor, get_api_3_data),
        loop.run_in_executor(executor, get_api_4_data)
    ]
    r1, r2, r3, r4 = await asyncio.gather(*tasks)
    return [r1, r2, r3, r4]

不过这种情况下，utilities.py里的async函数其实没必要留着，直接改成纯同步的更简洁。

注意点 #

• 线程池的max_workers可以按需调，一般设成要并行的任务数就行。
• 要是第三方库是CPU密集型任务，线程池提升效果有限，这时候可以试试ProcessPoolExecutor，但进程池开销大，适合纯CPU运算的场景。
• FastAPI基于ASGI，用线程池不会堵整个服务，事件循环会自动调度其他请求。

内容的提问来源于stack exchange，提问作者Abhishek Sachan