如何在FastAPI+SQLAlchemy中并行执行多查询与异步函数？

场景一：独立串行查询的并行改造 #

原来的串行查询会依次等待每个查询完成，总耗时是所有查询时间的总和。针对IO密集型的数据库查询，用线程池就能实现并行执行，大幅缩短总耗时：

from concurrent.futures import ThreadPoolExecutor

def execute_query():
    return cursor.query(Model).all()

# 初始化线程池，并行执行两个查询
with ThreadPoolExecutor(max_workers=2) as executor:
    future1 = executor.submit(execute_query)
    future2 = executor.submit(execute_query)
    # 获取两个查询的结果
    result_set1 = future1.result()
    result_set2 = future2.result()

两个查询会同时发起，总耗时基本等于单个查询的执行时间（线程调度开销可忽略）。

场景二：循环调用函数的并行优化 #

你写的asynFunc其实是同步函数，循环里逐个调用会串行等待。要让这10次调用并行执行，有两种可行方案：

方案1：线程池批量执行 #

适合不想改动原有同步查询代码的场景，直接用线程池批量提交任务：

from concurrent.futures import ThreadPoolExecutor

def sync_db_query():
    return cursor.query(Model).all()

with ThreadPoolExecutor(max_workers=5) as executor:
    # 一次性提交10个查询任务，自动并行执行
    responses = list(executor.map(sync_db_query, range(10)))

executor.map会自动分配任务到线程池，最后返回按顺序排列的结果列表。

方案2：改用真正的异步实现（需异步数据库驱动） #

如果你的项目基于异步框架（比如FastAPI），建议换成异步数据库驱动（比如SQLAlchemy异步版、asyncpg），再用asyncio.gather实现并行：

import asyncio
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy import select

# 初始化异步数据库连接
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
async_session = AsyncSession(engine)

async def async_db_query():
    async with async_session.begin():
        result = await async_session.execute(select(Model))
        return result.scalars().all()

async def run_parallel_queries():
    # 并行执行10个异步查询任务
    responses = await asyncio.gather(*[async_db_query() for _ in range(10)])
    return responses

# 启动异步任务
asyncio.run(run_parallel_queries())

注意：必须确保所有数据库操作都是异步的，否则无法实现真正的并行。

内容的提问来源于stack exchange，提问作者Deep Kumar Singh