FastAPI低延迟推理API的Azure Redis并发查询性能优化求助

我用FastAPI构建低延迟模型推理API，采用Azure Redis Cache标准版获取特征，ONNX模型做快速推理，aioredis实现Redis并发读取。单请求整体耗时70-80ms，但并发请求超10个时，Redis数据获取耗时突破400ms，且随并发数线性增长，高并发下甚至超900ms。目标是300并发请求/秒时总耗时低于300ms，500并发下仍维持低延迟，求优化方案。

当前实现代码 #

Redis数据获取模块 #

import numpy as np
import json
from ..Helpers.helper import curt_giver, milsec_calc
import aioredis

r = aioredis.from_url("redis://user:host", decode_responses=True)

async def get_user(user: list) -> str:
    user_data = await r.get(user)
    return user_data

async def get_products(product: list) -> list:
    product_data = await r.mget(product)
    return product_data

async def get_features(inputs: dict) -> list:
    
    st = curt_giver()
    user_data = await get_user(inputs['userId'])
    online_user_data = [json.loads(json.loads(user_data))]
    end = curt_giver()
    print("Time to get user features: ", milsec_calc(st,end))
    
    st = curt_giver()
    product_data = await get_products(inputs['productIds'])
    online_product_data = []
    for i in product_data:
        online_product_data.append(json.loads(json.loads(i)))
    end = curt_giver()
    print("Time to get product features: ", milsec_calc(st,end))

    user_outputs = np.asarray(online_user_data,dtype=object)
    product_outputs = np.asarray(online_product_data,dtype=object)
    output = np.concatenate([np.concatenate([user_outputs]*product_outputs.shape[0]), product_outputs], axis=1)
    return output.tolist()

FastAPI主模块 #

from fastapi import FastAPI
from v1.redis_conn.get_features import get_features
from model_scoring.score_onnx import score_features
from v1.post_processing.sort_results import sort_results
from v1.api_models.input_models import Ranking_Input
from v1.api_models.output_models import Ranking_Output
from v1.Helpers.helper import curt_giver, milsec_calc
import numpy as np

app = FastAPI()

# Sending user and product ids through body, 
# Hence a POST request is well suited for this, GET has unexpected behaviour
@app.post("/predict", response_model = Ranking_Output)
async def rank_products(inp_req: Ranking_Input):
    beg = curt_giver()
    reqids = inp_req.dict()
    st = curt_giver()
    features = await get_features(reqids)
    end = curt_giver()

    print("Total Redis duration ( user + products fetch): ", milsec_calc(st,end))

    data = np.asarray(features,dtype=np.float32,order=None)
    
    st = curt_giver()
    scores = score_features(data)
    end = curt_giver()

    print("ONNX model duration: ", milsec_calc(st,end))

    Ranking_results = sort_results(scores, list(reqids["productIds"]))
    end = curt_giver()
    print("Total time for API: ",milsec_calc(beg,end))
    resp_json = {
        "requestId": inp_req.requestId,
        "ranking": Ranking_results,
        "zipCode": inp_req.zipCode
    }

    return resp_json    

计时数据 #

单请求（单位：毫秒） #

Time to get user features:  1
Time to get product features:  47
Total Redis duration ( user + products fetch):  53
ONNX model duration:  2
Total time for API:  60

10+并发请求（单位：毫秒） #

Time to get user features:  151
Time to get user features:  150
Time to get user features:  151
Time to get user features:  52
Time to get user features:  51
Time to get product features:  187
Total Redis duration ( user + products fetch):  433
ONNX model duration:  2
Total time for API:  440
INFO:     127.0.0.1:60646 - "POST /predict HTTP/1.0" 200 OK
Time to get product features:  239
Total Redis duration ( user + products fetch):  488
ONNX model duration:  2
Total time for API:  495
INFO:     127.0.0.1:60644 - "POST /predict HTTP/1.0" 200 OK
Time to get product features:  142
Total Redis duration ( user + products fetch):  297
ONNX model duration:  2
Total time for API:  303
INFO:     127.0.0.1:60648 - "POST /predict HTTP/1.0" 200 OK
Time to get product features:  188
Total Redis duration ( user + products fetch):  342
ONNX model duration:  2
Total time for API:  348

优化方案 #

1. Redis连接池优化 #

当前默认连接池大小不足，高并发下会出现连接排队。显式指定更大的连接池（需匹配Azure Redis实例的连接数配额）：

r = aioredis.from_url(
    "redis://user:host",
    decode_responses=True,
    max_connections=500  # 建议大于目标并发数
)

2. 并行化Redis查询 #

将用户和产品的Redis查询从串行改为并行，减少总等待时间：

import asyncio

async def get_features(inputs: dict) -> list:
    # 并行发起两个查询任务
    user_task = get_user(inputs['userId'])
    product_task = get_products(inputs['productIds'])
    
    st = curt_giver()
    user_data, product_data = await asyncio.gather(user_task, product_task)
    end = curt_giver()
    print("Total parallel Redis fetch time: ", milsec_calc(st,end))

    online_user_data = [json.loads(json.loads(user_data))]
    online_product_data = [json.loads(json.loads(i)) for i in product_data]

    # 后续数组处理逻辑不变
    user_outputs = np.asarray(online_user_data,dtype=object)
    product_outputs = np.asarray(online_product_data,dtype=object)
    output = np.concatenate([np.concatenate([user_outputs]*product_outputs.shape[0]), product_outputs], axis=1)
    return output.tolist()

3. 减少JSON反序列化开销 #

当前双层JSON序列化/反序列化存在冗余，改为单层或二进制序列化（如MsgPack）：

• 存储时用单层json.dumps替代双层，或用msgpack.packb序列化数据
• 读取时对应改为单层json.loads或msgpack.unpackb解析

4. Azure Redis Cache配置升级 #

• 切换至高级层：提供更高吞吐量、更低延迟，支持更大连接数和分片功能，适配高并发场景
• 启用本地缓存：将热点特征缓存到应用服务器本地，减少Redis请求次数
• 调整过期策略：确保高频访问数据不被频繁驱逐，降低缓存失效带来的额外负载

5. 服务器运行参数优化 #

用Uvicorn启动时，使用高性能异步组件和足够的工作进程：

uvicorn main:app --workers 4 --loop uvloop --http httptools

uvloop和httptools能显著提升高并发下的异步处理效率。

6. 请求批处理（可选） #

若业务允许，合并多个请求的Redis查询为批量操作，比如用mget批量获取多个用户数据，减少连接开销。

7. 监控与瓶颈定位 #

• 通过Azure Portal监控Redis的CPU使用率、连接数、命令延迟，确认是否为Redis实例性能瓶颈
• 排查aioredis命令排队情况，定位慢查询

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