Flask后端处理15k-20k条JSON数据性能优化及防崩溃方案咨询

一、数据库操作优化（解决耗时过长问题） #

1. 批量查询已存在记录，避免N+1查询 #

先提取所有待导入数据的id，一次性查询数据库中已存在的ID集合，遍历数据时直接判断是否在集合中，替代逐条查询：

# 提取所有待导入的主数据ID
all_main_ids = [item['id'] for item in json_data]
# 批量查询已存在的ID
existing_main_ids = {obj.id for obj in MainModel.query.filter(MainModel.id.in_(all_main_ids)).all()}

# 遍历数据时直接判断
for item in json_data:
    if item['id'] not in existing_main_ids:
        # 处理新增逻辑

关联数据（如related字段）同理，先批量查询已存在的related.id，再批量处理新增。

2. 分批次提交事务，降低内存与数据库压力 #

不要一次性将20k条数据加入Session后提交，拆分小批次（如每1000条为一批）提交，减少Session内存占用与数据库锁时长：

batch_size = 1000
for i in range(0, len(json_data), batch_size):
    batch = json_data[i:i+batch_size]
    for item in batch:
        if item['id'] not in existing_main_ids:
            new_obj = MainModel(id=item['id'], name=item['name'])
            db.session.add(new_obj)
            # 处理关联数据的新增（同样批量处理）
    try:
        db.session.commit()
    except Exception as e:
        db.session.rollback()
        # 记录错误日志，标记该批次失败
    finally:
        db.session.remove()  # 清空Session，释放内存

3. 使用原生批量插入API #

利用SQLAlchemy的bulk_insert_mappings或bulk_save_objects替代逐条add，大幅提升插入效率：

# 筛选出需要新增的主数据
new_main_items = [item for item in json_data if item['id'] not in existing_main_ids]
# 批量插入主数据
db.session.bulk_insert_mappings(MainModel, new_main_items)
db.session.commit()

关联数据也可采用同样方式批量插入。

二、API负载控制（解决502/503错误） #

1. 异步处理请求，释放Flask进程 #

使用Celery等异步任务队列，将数据导入任务放入后台执行，前端仅获取任务ID，后续通过查询接口获取进度：

# 定义异步任务
@celery.task
def import_data_task(json_data):
    # 这里写批量导入的逻辑（上述优化后的代码）
    return {"success": len(new_items), "failed": 0}

# Flask接口
@app.route('/import', methods=['POST'])
def import_data():
    json_data = request.get_json()
    task = import_data_task.delay(json_data)
    return jsonify({"task_id": task.id})

# 任务查询接口
@app.route('/task/<task_id>')
def check_task(task_id):
    task = import_data_task.AsyncResult(task_id)
    if task.state == 'SUCCESS':
        return jsonify({"state": task.state, "result": task.result})
    return jsonify({"state": task.state})

2. 拆分前端请求，减少单次数据量 #

要求前端将20k条数据拆分为多个小批次（如每1k条发送一次请求），避免单次请求数据过大导致超时或内存溢出。

3. 限流与超时设置 #

使用Flask-Limiter对API请求进行限流，防止短时间内大量请求压垮服务器；同时设置合理的请求超时时间：

from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

limiter = Limiter(get_remote_address, app=app)

@app.route('/import', methods=['POST'])
@limiter.limit("5 per minute")  # 限制每分钟最多5次请求
def import_data():
    # 接口逻辑

同时在反向代理（如Nginx）中设置合理的proxy_read_timeout，避免因处理时间过长被判定为超时返回502。

4. 横向扩展与资源扩容 #

• 部署多个Flask实例，通过Nginx做负载均衡，分散请求压力；
• 根据服务器负载情况，适当增加CPU、内存资源，确保数据库与应用服务器有足够能力处理批量任务。

三、稳定性增强 #

1. 事务与错误处理 #

每个批次提交时添加异常捕获，出错则回滚当前批次，避免部分失败导致全量数据异常，同时记录详细错误日志便于排查。

2. 幂等性保障 #

在数据库中给id字段添加唯一约束，即使重复提交相同数据，数据库也会自动拒绝重复插入；或前端请求携带唯一请求ID，后端记录已处理的请求ID，避免重复执行。

3. 监控与日志 #

记录每个批次的处理时间、成功/失败数量，监控服务器CPU、内存、数据库连接数等指标，及时发现性能瓶颈或异常。

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