Flask+uWSGI下如何限制同一用户重复请求？

你的核心问题是竞态条件：并发请求同时检查session中的secret_variable，都发现为空后，会同时执行耗时的perform_secret_request。而uwsgi.lock()是进程级锁，多进程/线程环境下无法跨实例同步，所以失效。下面给出几种靠谱的解决办法：

一、Redis分布式锁（推荐） #

利用Redis的原子性操作实现跨进程/线程的全局锁，确保同一用户同一时间只有一个请求能进入临界区。

代码示例 #

import redis
from your_app import get_current_session, perform_secret_request

# 初始化Redis连接（建议做成单例）
redis_client = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)

def secret_route_handler(user):
    session_id = user.get('current_session_id')
    lock_key = f"user_request_lock:{session_id}"
    
    # 尝试获取锁：nx=True表示只有key不存在时才设置，ex=30是锁的过期时间（要比perform_secret_request最长耗时久）
    lock_acquired = redis_client.set(lock_key, "locked", nx=True, ex=30)
    if not lock_acquired:
        return {'error': '请求过于频繁，请稍后再试'}
    
    try:
        current_session = get_current_session(session_id)
        secret_var = current_session.get('secret_variable')
        if secret_var:
            return {'error': '用户已持有secret变量'}
        
        # 执行耗时操作
        secret_var = perform_secret_request()
        current_session.add_values({'secret_variable': secret_var})
    finally:
        # 无论成功失败，都释放锁
        redis_client.delete(lock_key)
    
    return {'data': secret_var}

关键注意点 #

• 锁的过期时间必须大于perform_secret_request的最长执行时间，防止进程意外挂掉导致锁永久占用。
• 用try-finally确保锁一定会被释放，避免死锁。

二、数据库行锁（适合已用数据库存储session的场景） #

如果你的session存在关系型数据库中，可以在查询session时加行锁，强制并发请求串行处理。

代码示例（以SQLAlchemy为例） #

from sqlalchemy import select
from sqlalchemy.orm import Session
from your_app.models import SessionModel, perform_secret_request

def secret_route_handler(user, db: Session):
    session_id = user.get('current_session_id')
    
    # 开启事务并对session行加排他锁，其他请求会阻塞直到锁释放
    with db.begin():
        current_session = db.execute(
            select(SessionModel)
            .where(SessionModel.id == session_id)
            .with_for_update()  # 加行锁
        ).scalar_one_or_none()
        
        if not current_session:
            return {'error': '会话不存在'}
        
        if current_session.secret_variable:
            return {'error': '用户已持有secret变量'}
        
        secret_var = perform_secret_request()
        current_session.secret_variable = secret_var
        db.commit()
    
    return {'data': secret_var}

关键注意点 #

• 仅适用于关系型数据库（MySQL、PostgreSQL等），NoSQL数据库需要对应行锁/文档锁机制。
• 锁的粒度是单用户session行，不会影响其他用户的请求。

三、Flask-Limiter限流扩展（适合简单场景） #

如果你的需求是固定时间窗口内同一用户只能请求一次，可以用Flask-Limiter快速实现，底层支持Redis存储来适配多进程/线程。

代码示例 #

from flask import Flask
from flask_limiter import Limiter
from your_app import get_current_user, get_current_session, perform_secret_request

app = Flask(__name__)
# 用session_id作为用户标识
def get_user_identifier():
    user = get_current_user()
    return user.get('current_session_id')

limiter = Limiter(
    key_func=get_user_identifier,
    app=app,
    storage_uri="redis://localhost:6379/0",  # 用Redis存储限流数据
)

@app.route('/secret')
@limiter.limit("1 per 30 seconds")  # 30秒内仅允许1次请求
def secret_endpoint():
    user = get_current_user()
    session_id = user.get('current_session_id')
    current_session = get_current_session(session_id)
    
    secret_var = current_session.get('secret_variable')
    if secret_var:
        return {'error': '用户已持有secret变量'}
    
    secret_var = perform_secret_request()
    current_session.add_values({'secret_variable': secret_var})
    
    return {'data': secret_var}

关键注意点 #

• 该方案是基于请求次数限流，而非检查session变量。如果需要等待session变量被移除后才能再次请求，还是前两种方案更合适。

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