Flask生产环境下如何避免高并发导致的重复数据处理问题

你的问题根源是竞态条件：多个请求同时执行find_one检查，此时都未找到对应u_id的文档，随后同时进入插入逻辑，最终导致重复数据。下面给出两种无需依赖唯一索引的可行方案：

方案一：使用MongoDB原子操作find_one_and_update #

利用MongoDB的原子操作将"检查+插入"合并为一步，从根本上避免竞态。通过$setOnInsert确保仅在文档不存在时才插入字段，upsert=True控制不存在则创建：

@app.route('/process_data/<u_id>')
def process_dt_(u_id):
    # 原子执行：仅当u_id不存在时插入新文档，存在则不做修改
    result = db_collection.find_one_and_update(
        {'u_id': u_id},
        {'$setOnInsert': {'key': 'value'}},
        upsert=True,
        return_document=True
    )
    
    # 通过upserted_id判断是否是新插入的文档
    if hasattr(result, 'upserted_id') and result.upserted_id is not None:
        return 'OK'
    else:
        return 'has already been processed', 403

优势：无需额外依赖组件，性能高，完全由MongoDB保证操作原子性，是最推荐的方案。

方案二：分布式锁（基于Redis） #

如果业务逻辑更复杂，无法用单一MongoDB原子操作覆盖，可以引入分布式锁，保证同一u_id的请求串行处理：

import redis
from flask import Flask

app = Flask(__name__)
# 初始化Redis客户端（根据你的配置调整参数）
redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

def acquire_lock(u_id, lock_timeout=10):
    # 尝试获取锁，nx=True表示仅当key不存在时设置，ex设置锁过期时间防止死锁
    return redis_client.set(f"lock:{u_id}", "locked", ex=lock_timeout, nx=True)

def release_lock(u_id):
    redis_client.delete(f"lock:{u_id}")

@app.route('/process_data/<u_id>')
def process_dt_(u_id):
    # 无法获取锁时，直接返回冲突
    if not acquire_lock(u_id):
        return 'processing or already processed', 403
    
    try:
        # 原有逻辑现在是串行执行，不会出现竞态
        check_exists = db_collection.find_one({'u_id': u_id})
        if not check_exists:
            db_collection.insert_one({
                'u_id': u_id,
                'key': 'value'
            })
            return 'OK'
        else:
            return 'has already been processed', 403
    finally:
        # 无论成功失败都释放锁
        release_lock(u_id)

注意事项：

• 锁的过期时间要设置合理，避免业务执行时间超过锁有效期导致的锁提前释放。
• 可以考虑引入锁续约机制，防止长耗时任务的锁过期。

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