Flask多API调用异步化：如何缩短应用加载时间？

针对你的场景，有两种主流方案可以实现多API调用的异步执行，分别适配不同的改造成本和性能需求：

方案一：线程池并行执行（低改造成本） #

你的现有API调用基于同步的requests库，最快的改造方式是用线程池把多个同步调用并行起来——IO密集型任务用线程池能有效减少等待时间，几乎不用修改原有函数逻辑。

示例代码如下：

from flask import Flask
import requests
from concurrent.futures import ThreadPoolExecutor

app = Flask(__name__)

# 保留你原有的API调用函数
def api_call_1():
    try:
        url = "https://example.com/api1"
        response = requests.get(url)
        response.raise_for_status()
        res = response.json()
        return res["key"]
    except (requests.RequestException, KeyError, TypeError, ValueError):
        return "Oops, there was an error!"

def api_call_2():
    try:
        url = "https://example.com/api2"
        response = requests.get(url)
        response.raise_for_status()
        res = response.json()
        return res["another_key"]
    except (requests.RequestException, KeyError, TypeError, ValueError):
        return "Oops, there was an error!"

@app.route("/")
def index():
    # 全局初始化线程池更高效，避免每次请求重复创建
    with ThreadPoolExecutor(max_workers=3) as executor:
        # 提交所有API调用任务
        future1 = executor.submit(api_call_1)
        future2 = executor.submit(api_call_2)
        # 获取任务结果，也可用as_completed处理先完成的任务
        result1 = future1.result()
        result2 = future2.result()
    
    return f"Result 1: {result1}, Result 2: {result2}"

if __name__ == "__main__":
    app.run(debug=True)

注意：max_workers可根据API调用数量调整，建议不超过CPU核心数的2-3倍，避免资源过载。

方案二：异步IO实现（更高性能） #

如果追求更优的性能，推荐改用异步HTTP库aiohttp配合Flask 2.0+支持的异步视图，基于协程实现无阻塞调用，线程开销更低。

首先安装依赖：

pip install aiohttp flask

改造后的代码示例：

from flask import Flask
import aiohttp
import asyncio

app = Flask(__name__)

# 全局复用ClientSession，避免每次请求重复创建
session = aiohttp.ClientSession()

# 将同步函数改造为异步协程
async def async_api_call_1():
    try:
        url = "https://example.com/api1"
        async with session.get(url) as response:
            response.raise_for_status()
            res = await response.json()
            return res["key"]
    except (aiohttp.ClientError, KeyError, TypeError, ValueError):
        return "Oops, there was an error!"

async def async_api_call_2():
    try:
        url = "https://example.com/api2"
        async with session.get(url) as response:
            response.raise_for_status()
            res = await response.json()
            return res["another_key"]
    except (aiohttp.ClientError, KeyError, TypeError, ValueError):
        return "Oops, there was an error!"

# Flask异步视图
@app.route("/async")
async def async_index():
    # 并发执行所有异步API调用，return_exceptions设为True可捕获异常并返回
    result1, result2 = await asyncio.gather(
        async_api_call_1(),
        async_api_call_2(),
        return_exceptions=False
    )
    return f"Async Result 1: {result1}, Async Result 2: {result2}"

# 应用关闭时清理session
@app.teardown_appcontext
async def close_session(exception):
    await session.close()

if __name__ == "__main__":
    app.run(debug=True)

两种方案对比 #

• 线程池方案：几乎无需修改原有API函数，快速实现并行，适合小范围改造场景。
• 异步IO方案：需要改造原有函数为协程，性能更优，适合高并发或大量API调用的场景。

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