如何为Flask的/build-task端点添加互斥锁，避免并发调用？

嘿，这个需求太典型了——长时间运行的API端点最怕重复触发，轻则浪费资源，重则导致数据混乱。我给你几个不同场景下的解决方案，你可以按需选择：

如果你的Flask是单进程运行（比如直接用flask run测试，或者生产环境单worker），最简单的方式就是用一个全局变量来标记任务是否在运行：

from flask import Flask, jsonify

app = Flask(__name__)
# 全局锁标记，初始为False表示任务未运行
is_build_running = False

@app.route('/build-task', methods=['POST'])
def build_task():
    global is_build_running
    if is_build_running:
        return jsonify({"error": "Build task is already running"}), 409
    
    try:
        is_build_running = True
        # 模拟30分钟的任务
        # time.sleep(1800)
        print("Starting long build task...")
        # 这里写你的实际任务逻辑
        return jsonify({"message": "Build task completed successfully"})
    finally:
        # 不管任务成功失败，都要重置锁
        is_build_running = False

if __name__ == '__main__':
    app.run()

⚠️ 注意：这个方案只适用于单进程场景，如果用了多worker（比如Gunicorn开多个进程），全局变量是进程隔离的，每个worker都有自己的is_build_running，锁就失效了。

如果是生产环境多进程或者分布式部署，就得用外部共享存储来存锁状态，比如Redis或者数据库。这里举Redis的例子，因为它的SETNX（Set if Not Exists）命令天生适合做分布式锁：

首先安装Redis依赖：

pip install redis

然后修改代码：

from flask import Flask, jsonify
import redis
import time

app = Flask(__name__)
# 连接Redis，根据你的配置修改
r = redis.Redis(host='localhost', port=6379, db=0)
# 锁的key，自定义一个唯一标识
LOCK_KEY = "build_task_running"
# 锁的过期时间，设置比任务最长运行时间长一点，比如35分钟=2100秒
LOCK_EXPIRE = 2100

@app.route('/build-task', methods=['POST'])
def build_task():
    # 尝试获取锁：如果key不存在就设置，返回True；否则返回False
    acquired = r.set(LOCK_KEY, "1", ex=LOCK_EXPIRE, nx=True)
    if not acquired:
        return jsonify({"error": "Build task is already running"}), 409
    
    try:
        print("Starting long build task...")
        # 模拟30分钟任务
        # time.sleep(1800)
        # 你的实际任务逻辑
        return jsonify({"message": "Build task completed successfully"})
    finally:
        # 任务结束后主动释放锁
        r.delete(LOCK_KEY)

if __name__ == '__main__':
    app.run()

这个方案的好处是不管多少个worker或者多少台服务器，都能共享锁状态，而且设置了过期时间，就算任务意外崩溃（比如服务器断电），锁也会自动过期，不会一直卡死。

如果你已经在用Flask-Caching扩展，也可以用它来实现锁，代码更简洁：

先安装：

pip install flask-caching

代码示例：

from flask import Flask, jsonify
from flask_caching import Cache
import time

app = Flask(__name__)
cache_config = {
    "CACHE_TYPE": "RedisCache",
    "CACHE_REDIS_HOST": "localhost",
    "CACHE_REDIS_PORT": 6379,
    "CACHE_REDIS_DB": 0
}
cache = Cache(app, config=cache_config)
LOCK_KEY = "build_task_running"
LOCK_EXPIRE = 2100

@app.route('/build-task', methods=['POST'])
def build_task():
    # 尝试获取缓存，不存在就设置并返回True
    if cache.get(LOCK_KEY):
        return jsonify({"error": "Build task is already running"}), 409
    
    try:
        cache.set(LOCK_KEY, "1", timeout=LOCK_EXPIRE)
        print("Starting long build task...")
        # time.sleep(1800)
        # 任务逻辑
        return jsonify({"message": "Build task completed successfully"})
    finally:
        cache.delete(LOCK_KEY)

if __name__ == '__main__':
    app.run()

额外提醒 #

• 一定要给锁设置合理的过期时间，比你的任务最长可能运行时间长一点，避免死锁。
• 任务执行过程中如果有异常，一定要在finally块里释放锁（或者依赖过期时间兜底）。
• 如果是非常关键的任务，还可以考虑给锁加唯一标识（比如用UUID），防止误释放其他进程的锁（不过一般简单场景用上面的方法足够了）。

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