FastAPI返回70万行数据响应耗时长 阻塞服务无法处理其他请求

返回阶段出现长达2分钟的服务阻塞，核心原因是FastAPI默认的响应序列化逻辑运行在asyncio主事件循环线程中。
你通过run_in_threadpool封装数据库查询逻辑，让IO密集的查询操作跑在线程工作池，因此50秒查询阶段不会占用事件循环，其他请求可以正常处理。但直接返回Python字典时，FastAPI会在主事件循环内同步执行「字典转JSON字符串」的CPU密集操作，70万行数据的序列化过程会完全占住事件循环线程，导致所有待处理的请求协程无法被调度，对外表现为服务完全卡死。
由于业务限制无法使用分页/limit限制返回数据量，以下是可直接落地的解决方案，按改造成本从低到高排序： #

方案1：替换高性能JSON库 + 序列化逻辑移至线程池 #

该方案不需要调整接口返回格式，改造成本最低，见效最快：

• 用orjson替代Python标准库json做序列化，针对大数据量场景，其序列化速度是标准库的3-5倍，内存占用也更低
• 不要直接返回原始字典让框架自动序列化，而是把序列化操作也提交到线程池执行，待序列化完成后直接返回构造好的二进制响应，彻底避免CPU密集操作阻塞主循环
  参考代码：

import orjson
from fastapi import FastAPI, Response
from fastapi.concurrency import run_in_threadpool

app = FastAPI()

# 原有PG查询逻辑保持不变
def get_data_from_pgsql(data):
    # 你的数据库查询实现，建议直接返回字典结构结果，避免ORM对象额外转换开销
    pass

@app.get("/request")
async def request_db(data):
    dict_of_result = await run_in_threadpool(get_data_from_pgsql, data)
    # 序列化操作移到线程池执行，不占用主事件循环
    json_content = await run_in_threadpool(orjson.dumps, dict_of_result)
    # 直接返回已序列化完成的响应，框架不会重复做序列化处理
    return Response(content=json_content, media_type="application/json")

改完后原有2分钟的阻塞会直接消除：一方面序列化本身耗时会被压缩到原来的1/3~1/5，另一方面序列化全程跑在工作线程，主事件循环始终可以正常调度其他请求。 #

方案2：使用StreamingResponse流式返回数据 #

如果要进一步降低服务内存占用、缩短接口首包响应时间，可以采用流式响应方案，不需要等全量数据查询、序列化完成后再一次性返回，而是分批拉取数据、分批序列化、分批传输给客户端：

• 数据库查询改用PG服务端游标，每次只拉取小批量数据，避免70万行数据全量加载到内存
• 流式生成JSON内容的逻辑全程跑在工作线程，不会阻塞主事件循环
  参考代码：

import orjson
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

# 你的PG连接获取逻辑
def get_pg_conn():
    pass

def batch_fetch_data(data):
    # 用命名游标启用PG服务端游标，不会一次性加载全量结果
    conn = get_pg_conn()
    cursor = conn.cursor(name="fastapi_stream_cursor")
    cursor.execute("你的查询SQL语句", data)
    try:
        while True:
            # 每次拉取2000行，可根据单条数据大小调整批次大小
            batch = cursor.fetchmany(2000)
            if not batch:
                break
            yield batch
    finally:
        cursor.close()
        conn.close()

def stream_json_content(data):
    # 同步生成器，Starlette会自动将其放到线程池执行，不阻塞主循环
    yield b"["
    first_row = True
    for batch in batch_fetch_data(data):
        for row in batch:
            if not first_row:
                yield b","
            first_row = False
            yield orjson.dumps(dict(row))
    yield b"]"

@app.get("/request")
async def request_db(data):
    return StreamingResponse(
        stream_json_content(data),
        media_type="application/json"
    )

该方案的优势是内存占用极低，无论返回多少行数据，服务内存中仅缓存当前批次的数千条记录，不会出现全量数据堆积在内存的问题，且全程没有长时间占用事件循环的逻辑，服务可用性更高。 #

额外优化建议 #

• 若使用SQLAlchemy等ORM框架，查询时建议直接调用mappings()方法返回字典结构结果，跳过ORM实例对象转字典的额外开销
• 线程池大小建议按服务CPU核数调整，CPU密集型任务的工作线程数设置为CPU核数+1即可，避免过多线程上下文切换拖慢处理速度
• 若返回数据不需要在Python层做二次加工，可以直接在PG查询时用json_agg、row_to_json等内置函数直接生成JSON结果，跳过Python层序列化步骤，进一步压缩耗时

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