如何优化PostgreSQL查询？FastAPI多查询场景代码优化问询

1. 用任务配置列表简化重复代码 #

把所有fetch_one的参数整理成配置列表，动态生成任务，避免重复写fetch_one调用：

# 定义所有查询任务的配置元组：(查询标签, 别名, 参数)
query_configs = [
    ("mutualfund_overall_summary", overall_summary_alias, kwargs_ffs),
    ("mutualfund_category_summary", overall_summary_alias, kwargs_ffs),
    ("mutualfund_account_details", overall_summary_alias, kwargs_ffs),
    ("nps_overall_summary", overall_summary_alias, kwargs_ffs),
    ("ppf_account_details", overall_summary_alias, kwargs_ffs),
    ("equity_overall_summary", overall_summary_alias, kwargs_ffs),
    ("equity_account_details", overall_summary_alias, kwargs_ffs),
    ("rd_overall_summary", fd_rd_alias, kwargs_ffs),
    ("rd_account_details", fd_rd_alias, kwargs_ffs),
    ("fd_summary", fd_rd_alias, kwargs_ffs),
    ("fd_account_details", fd_rd_alias, kwargs_ffs),
    ("insurance_overall_summary", insurance_overall_summary_alias, kwargs_ffs),
    ("bureau_overall_summary", bureau_overall_summary_alias, kwargs_ffs),
    ("bureau_loan_account_details", bureau_overall_summary_alias, kwargs_ffs),
    ("bureau_creditcard_account_details", bureau_overall_summary_alias, kwargs_ffs),
    ("bureau_loan_category_summary", bureau_overall_summary_alias, kwargs_ffs),
    ("ffs_category_summary", ffs_category_summary_alias, ffs_category_summary_kwargs),
]

# 生成fetch_one任务列表
fetch_tasks = [fetch_one(tag, alias, params) for tag, alias, params in query_configs]
# 添加缓存日期查询任务
fetch_tasks.append(get_cached_engine_date("ffs"))

# 批量执行所有任务
results = await asyncio.gather(*fetch_tasks)

# 按原顺序解包到变量（和原代码变量一一对应）
(mf_overall_summary, mf_category_summary, mf_account_details, nps, ppf, 
 equity_overall, equity_account, rd_overall, rd_account, fd_overall, fd_account, 
 insurance, bureau_overall, bureau_loan_account, bureau_creditcard, 
 bureau_loan_category, overall_data, ffs_engine_date) = results

这种方式维护起来更方便，新增/删除查询只需修改query_configs列表，不用重复写fetch_one。

2. 用字典映射变量与任务（更灵活） #

如果后续不需要单独的变量名，或者想更清晰地管理结果，可以用字典映射：

# 用字典关联变量名和任务配置
task_mapping = {
    "mf_overall_summary": ("mutualfund_overall_summary", overall_summary_alias, kwargs_ffs),
    "mf_category_summary": ("mutualfund_category_summary", overall_summary_alias, kwargs_ffs),
    "mf_account_details": ("mutualfund_account_details", overall_summary_alias, kwargs_ffs),
    "nps": ("nps_overall_summary", overall_summary_alias, kwargs_ffs),
    "ppf": ("ppf_account_details", overall_summary_alias, kwargs_ffs),
    "equity_overall": ("equity_overall_summary", overall_summary_alias, kwargs_ffs),
    "equity_account": ("equity_account_details", overall_summary_alias, kwargs_ffs),
    "rd_overall": ("rd_overall_summary", fd_rd_alias, kwargs_ffs),
    "rd_account": ("rd_account_details", fd_rd_alias, kwargs_ffs),
    "fd_overall": ("fd_summary", fd_rd_alias, kwargs_ffs),
    "fd_account": ("fd_account_details", fd_rd_alias, kwargs_ffs),
    "insurance": ("insurance_overall_summary", insurance_overall_summary_alias, kwargs_ffs),
    "bureau_overall": ("bureau_overall_summary", bureau_overall_summary_alias, kwargs_ffs),
    "bureau_loan_account": ("bureau_loan_account_details", bureau_overall_summary_alias, kwargs_ffs),
    "bureau_creditcard": ("bureau_creditcard_account_details", bureau_overall_summary_alias, kwargs_ffs),
    "bureau_loan_category": ("bureau_loan_category_summary", bureau_overall_summary_alias, kwargs_ffs),
    "overall_data": ("ffs_category_summary", ffs_category_summary_alias, ffs_category_summary_kwargs),
}

# 生成任务列表并记录变量顺序
tasks = []
variable_order = list(task_mapping.keys())
for var_name, (tag, alias, params) in task_mapping.items():
    tasks.append(fetch_one(tag, alias, params))
# 添加特殊任务
tasks.append(get_cached_engine_date("ffs"))
variable_order.append("ffs_engine_date")

# 执行任务并转成字典结果
results = await asyncio.gather(*tasks)
result_dict = dict(zip(variable_order, results))

# 后续使用时直接从字典取值
# mf_overall_summary = result_dict["mf_overall_summary"]

这种方式可以避免维护过长的变量解包语句，结果更易管理。

1. 索引优化 #

• 为查询中常用的过滤条件、JOIN连接字段创建B-tree索引（适合等值、范围查询）；如果涉及模糊匹配或JSON字段，可使用GIN/GIST索引。
• 避免过度创建索引：索引会加速查询，但会增加写入（INSERT/UPDATE/DELETE）的开销，只给高频查询的字段加索引。
• 定期用REINDEX整理碎片化的索引。

2. 查询语句优化 #

• 避免SELECT *，只返回业务需要的字段，减少数据传输量。
• 优化JOIN逻辑：确保连接字段有索引，避免笛卡尔积；子查询嵌套过深时，改用CTE（WITH子句）或JOIN重构，让查询优化器更好地生成执行计划。
• 对于批量查询，若多个查询逻辑关联且数据量不大，可尝试合并成单个查询（用UNION ALL或多表关联），减少数据库连接开销。

3. 连接池与缓存优化 #

• 异步场景下，合理设置数据库连接池大小（建议根据CPU核心数调整，比如核心数*2），避免连接过多导致数据库资源耗尽。
• 对不频繁更新的查询结果（如get_cached_engine_date），用Redis等缓存工具缓存，减少重复查询数据库的次数。

4. 数据库配置调优 #

• 调整postgresql.conf中的关键参数：
  • shared_buffers：建议设为服务器内存的25%，提升数据库缓存能力。
  • work_mem：单个查询可用的内存，若查询涉及排序、哈希JOIN，适当调大避免频繁磁盘读写。
  • effective_cache_size：设为内存的50%-75%，帮助查询优化器估算缓存命中率。
• 根据业务场景调整max_connections，避免连接数超限。

5. 分析与监控 #

• 用EXPLAIN ANALYZE分析慢查询的执行计划，定位全表扫描、排序开销大等瓶颈。
• 开启PostgreSQL的慢查询日志（设置log_min_duration_statement），跟踪耗时较长的查询。
• 对于超大表，考虑按时间、业务维度做分区表，减少查询时扫描的数据量。

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