请求列表扩容时FastAPI并行请求响应变慢的问题排查与优化

一、mktlist条目增多时响应变慢的原因 #

1. 单Worker异步调度开销累积：单worker的Uvicorn默认采用单线程事件循环，当并行请求数量增加到10条左右时，事件循环需要处理更多的IO回调、任务切换操作。即使外部API响应极快，这些调度和回调的额外开销会被累加，导致整体响应耗时上升。而1-2条请求时，调度开销占比极低，所以耗时接近外部API。
2. TCP连接池限制：aiohttp默认的连接池配置（如单主机连接数限制）可能导致并行请求时出现连接等待。当发起10条并行请求，如果连接池没有足够的空闲连接，会产生连接建立或等待复用的额外耗时。
3. ECS资源瓶颈：单worker的ECS容器如果CPU、内存配额不足，当并行处理多个外部请求时，事件循环的调度、请求/响应序列化等操作会出现资源竞争，导致任务执行延迟。
4. 请求处理的同步开销放大：mktlist条目越多，FastAPI的输入参数解析、响应数据组装等同步操作的工作量越大，这些操作会阻塞事件循环，拖慢整体响应速度。

二、并行请求的优化方法 #

1. 优化aiohttp连接池配置 #

显式配置TCP连接池，提升连接复用率，减少握手开销：

from aiohttp import ClientSession, TCPConnector

# 增大连接池大小，开启长连接
connector = TCPConnector(
    limit=50,
    limit_per_host=50,
    keepalive_timeout=30,
    enable_cleanup_closed=True
)

async def fetch_external_api(urls):
    async with ClientSession(connector=connector) as session:
        tasks = [session.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        return [await resp.json() for resp in responses]

2. 调整Uvicorn运行配置 #

• 启用多Worker模式：根据ECS容器的CPU核数设置Worker数量（建议为2*CPU核数+1），利用多进程分摊负载：

  uvicorn main:app --workers 3 --host 0.0.0.0 --port 8000
  

• 替换为uvloop事件循环：uvloop比默认asyncio事件循环性能更高，能提升异步任务调度效率：

  uvicorn main:app --workers 2 --loop uvloop --host 0.0.0.0 --port 8000
  

3. 控制并行请求并发数 #

使用asyncio.Semaphore限制同时发起的外部请求数量，避免事件循环过载：

import asyncio
from aiohttp import ClientSession

semaphore = asyncio.Semaphore(10)  # 限制最大并发数为10

async def fetch_single_url(session, url):
    async with semaphore:
        async with session.get(url) as resp:
            return await resp.json()

async def batch_fetch(urls):
    async with ClientSession() as session:
        tasks = [fetch_single_url(session, url) for url in urls]
        return await asyncio.gather(*tasks)

4. 扩容ECS资源 #

检查ECS容器的CPU和内存配额，若使用率接近上限，提升资源配置（比如从0.5vCPU/1GB内存调整为1vCPU/2GB内存），确保有足够资源处理并行任务。

5. 减少同步操作阻塞 #

将耗时的同步操作（如复杂数据转换）移到线程池执行，避免阻塞事件循环：

from concurrent.futures import ThreadPoolExecutor
import asyncio

executor = ThreadPoolExecutor(max_workers=4)

def heavy_data_processing(raw_data):
    # 耗时的同步数据处理逻辑
    return processed_data

async def process_responses(raw_responses):
    loop = asyncio.get_event_loop()
    tasks = [loop.run_in_executor(executor, heavy_data_processing, data) for data in raw_responses]
    return await asyncio.gather(*tasks)

6. 添加响应缓存 #

对于更新不频繁的外部API响应，使用本地缓存（如functools.lru_cache）或分布式缓存（如Redis），减少重复请求：

from functools import lru_cache

# 注意：异步函数不能直接用lru_cache，可封装同步逻辑
@lru_cache(maxsize=100)
def cached_external_data(url):
    # 同步请求外部API（或调用异步转同步的逻辑）
    pass

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