性能调优

概述

Flask 应用的性能调优涉及 Gunicorn Worker 配置、数据库连接池、Redis 缓存策略、慢查询优化等多个层面。合理的调优能显著提升并发处理能力和响应速度。

性能调优原则

1. 先测量，后优化：用数据说话，避免盲目调参
2. 瓶颈优先：找到最大瓶颈，集中优化
3. 渐进式：每次只调整一个参数，观察效果

Gunicorn Worker 配置

Worker 数量

# 单 Worker（开发环境）
python app.py

# 多 Worker（生产环境）
gunicorn -w 4 -b 0.0.0.0:8021 app:app

Worker 数量公式

推荐 Worker 数量 = CPU 核心数 * 2 + 1

例如 4 核 CPU：workers = 4 * 2 + 1 = 9

查看 CPU 核心数：

# Linux
nproc

# Python
import multiprocessing; print(multiprocessing.cpu_count())

启动方式选择

方式	适用场景	特点
python app.py	开发调试	单进程，自动重载
gunicorn -w N	生产部署	多进程，无自动重载
gunicorn --threads N	生产部署	多线程模式

Gunicorn 配置示例

pip install gunicorn

gunicorn app:app \
    --workers 4 \
    --bind 0.0.0.0:8021 \
    --timeout 120 \
    --keep-alive 5 \
    --max-requests 1000 \
    --max-requests-jitter 50

max-requests 作用

--max-requests 1000 表示每个 Worker 处理 1000 个请求后自动重启，防止内存泄漏累积。--max-requests-jitter 50 添加随机抖动，避免所有 Worker 同时重启。

Gunicorn 配置文件

创建 gunicorn.conf.py 配置文件：

# gunicorn.conf.py
import multiprocessing

# 绑定地址
bind = "0.0.0.0:8021"

# Worker 数量
workers = multiprocessing.cpu_count() * 2 + 1

# Worker 类型
worker_class = "sync"

# 线程数（worker_class="gthread" 时生效）
threads = 2

# 超时时间
timeout = 120

# 保持连接
keepalive = 5

# 最大请求数（防止内存泄漏）
max_requests = 1000
max_requests_jitter = 50

# 日志
accesslog = "-"
errorlog = "-"
loglevel = "info"

使用配置文件启动：

gunicorn -c gunicorn.conf.py app:app

数据库连接池

SQLAlchemy 连接池配置（在 config/database.py 中）：

# 连接池配置
SQLALCHEMY_POOL_SIZE = 20       # 连接池大小
SQLALCHEMY_MAX_OVERFLOW = 10    # 超出池大小的最大连接数
SQLALCHEMY_POOL_TIMEOUT = 30    # 获取连接超时（秒）
SQLALCHEMY_POOL_RECYCLE = 3600  # 连接回收时间（秒）

连接池大小

连接池总连接数 = pool_size + max_overflow。需确保不超过数据库的 max_connections 限制，否则会报连接数超限错误。

MySQL 查看最大连接数：

SHOW VARIABLES LIKE 'max_connections';

连接池监控

# 获取连接池状态
from src.core.database import engine
pool = engine.pool
print(f"池大小: {pool.size()}")
print(f"已签出: {pool.checkedout()}")
print(f"空闲数: {pool.checkedin()}")
print(f"溢出数: {pool.overflow()}")

Redis 缓存优化

缓存策略

策略	适用场景	实现方式
读缓存	字典、配置等读多写少	查询时先读 Redis，未命中再查 DB
写失效	数据更新时	更新 DB 后删除 Redis 缓存
预热	服务启动时	启动后主动加载热点数据到 Redis

TTL 设置建议

# 字典数据：较长 TTL（变更频率低）
DICT_CACHE_TTL = 3600       # 1 小时

# 用户权限：中等 TTL（登录时刷新）
PERM_CACHE_TTL = 1800       # 30 分钟

# 验证码：短 TTL（一次性使用）
CAPTCHA_CACHE_TTL = 300     # 5 分钟

缓存预热

# 服务启动时预热缓存
def warmup_cache():
    """预热字典、配置等热点数据"""
    from src.common.utils.dict_util import refresh_dict_cache
    refresh_dict_cache()

慢查询优化

开启慢查询日志

MySQL：

SET GLOBAL slow_query_log = ON;
SET GLOBAL long_query_time = 1;
SET GLOBAL slow_query_log_file = '/var/log/mysql/slow.log';

常见优化手段

问题	优化方案
缺少索引	为 WHERE/ORDER BY 字段添加索引
全表扫描	使用 EXPLAIN 分析查询计划
N+1 查询	使用 joinedload/subqueryload 预加载关联数据
大分页	使用游标分页或延迟关联
SELECT *	只查询需要的字段

索引优化示例

# 模型定义中添加索引
class Article(BaseModel, BaseDB):
    __tablename__ = DB_PREFIX + "article"

    title = Column(String(255), index=True)        # 单列索引
    status = Column(Integer, index=True)
    category_id = Column(Integer, index=True)

    # 复合索引（在 __table_args__ 中定义）
    __table_args__ = (
        Index('idx_status_category', 'status', 'category_id'),
    )

分页查询优化

# 避免大分页：使用游标分页
# 差：OFFSET 100000 LIMIT 10（扫描大量数据）
# 好：WHERE id > last_id LIMIT 10（利用主键索引）

def paginate_after_id(self, after_id: int, limit: int = 10):
    """游标分页：基于上次最后一条记录的 ID"""
    return cls.model.query.filter(
        cls.model.id > after_id,
        cls.model.is_delete == 0
    ).order_by(cls.model.id.asc()).limit(limit).all()

同步架构优化

线程池配置

Flask 使用 Gunicorn 的 sync worker，每个 worker 处理一个请求。可以通过多 worker 和多线程提升并发：

# 多 worker + 多线程
gunicorn app:app \
    --workers 4 \
    --threads 2 \
    --worker-class gthread \
    --bind 0.0.0.0:8021

异步任务

对于耗时操作（如邮件发送、文件处理），使用 Celery 等异步任务队列：

from celery import Celery

celery = Celery('tasks', broker='redis://localhost:6379/0')

@celery.task
def send_email(to, subject, body):
    """异步发送邮件"""
    # 发送邮件逻辑
    pass

总结

性能调优的核心路径：Gunicorn 多 Worker 利用多核 CPU、数据库连接池匹配并发需求、Redis 缓存减少 DB 压力、索引和查询优化降低慢查询。建议先用监控数据定位瓶颈，再针对性调优，每次只调整一个参数并观察效果。