Python Flask：如何控制后台线程数量并处理请求过载

Hey there! Let's work through this problem together. Since you're already running Flask with threaded=True, we just need to add a concurrency limiter to cap active detection threads and reject overflow requests with a "server busy" message. Here's how to do it:

核心思路：使用线程信号量（Semaphore） #

Python的threading.Semaphore是控制并发线程数的完美工具——它维护着一组"许可"，每个线程必须获取许可才能执行任务，任务完成后释放许可。当所有许可都被占用时，新的请求会被直接拒绝。

完整实现代码 #

from flask import Flask, request, jsonify
import threading
import time  # 仅用于模拟耗时检测，实际替换为你的检测逻辑

app = Flask(__name__)

# 配置最大并发线程数（你可以设为4或5）
MAX_CONCURRENT_TASKS = 4
# 初始化信号量，设置许可数量
task_semaphore = threading.Semaphore(MAX_CONCURRENT_TASKS)

def run_object_detection(image_data):
    """模拟耗时的目标检测任务，替换成你的实际检测代码"""
    try:
        # 这里是你的目标检测逻辑，比如加载模型、处理图像、输出结果
        time.sleep(6)  # 模拟6秒的检测耗时
        return {"status": "success", "detected_objects": ["car", "pedestrian"]}
    finally:
        # 不管任务成功/失败，都必须释放信号量，避免许可泄漏
        task_semaphore.release()

@app.route('/api/detect', methods=['POST'])
def handle_detection_request():
    # 尝试获取信号量，不阻塞（立即返回结果）
    if not task_semaphore.acquire(blocking=False):
        # 无可用许可，返回服务器繁忙，HTTP 503是标准的服务不可用状态码
        return jsonify({"error": "服务器繁忙，请稍后重试"}), 503
    
    # 解析请求数据（根据你的实际请求格式调整，比如表单、二进制图像等）
    request_data = request.get_json()
    if not request_data or 'image_data' not in request_data:
        # 参数错误，记得释放信号量再返回
        task_semaphore.release()
        return jsonify({"error": "缺少必要的图像数据"}), 400
    
    # 启动后台线程执行检测任务，避免阻塞Flask的请求线程
    detection_thread = threading.Thread(
        target=run_object_detection,
        args=(request_data['image_data'],)
    )
    detection_thread.start()
    
    # 返回任务接受的响应（如果需要同步返回结果，参考下面的同步示例）
    return jsonify({"message": "检测任务已提交，正在后台处理"}), 202

# 可选：同步返回检测结果的路由（客户端需等待任务完成）
@app.route('/api/detect_sync', methods=['POST'])
def handle_sync_detection():
    if not task_semaphore.acquire(blocking=False):
        return jsonify({"error": "服务器繁忙，请稍后重试"}), 503
    
    try:
        request_data = request.get_json()
        if not request_data or 'image_data' not in request_data:
            return jsonify({"error": "缺少必要的图像数据"}), 400
        
        # 直接执行检测（会阻塞当前请求线程，直到任务完成）
        result = run_object_detection(request_data['image_data'])
        return jsonify(result), 200
    finally:
        # 确保信号量被释放
        task_semaphore.release()

if __name__ == '__main__':
    app.run(threaded=True, debug=False)

关键细节说明 #

• 信号量的释放：一定要用finally块确保信号量被释放，哪怕检测过程中抛出异常，也不会导致许可被永久占用（死锁）。
• HTTP状态码选择：返回503 Service Unavailable是符合HTTP规范的，客户端可以识别这是临时状态，适合重试。
• 异步vs同步处理：上面的示例提供了两种模式：
  • 异步模式：客户端提交请求后立即收到响应，检测任务在后台执行（适合耗时较长的任务）。
  • 同步模式：客户端等待检测完成后拿到结果（适合需要实时返回的场景，但会占用请求线程直到任务结束）。

生产环境注意事项 #

如果你用Gunicorn等生产级服务器部署：

• 如果是单进程多线程模式（比如gunicorn --workers=1 --threads=4 app:app），上面的信号量可以直接生效，总并发数和你设置的MAX_CONCURRENT_TASKS一致。
• 如果是多进程模式（workers>1），每个进程会有独立的信号量，这时候需要用分布式锁（比如Redis的Semaphore）来全局控制并发数，避免总线程数超过预期。

Hope this solves your problem! Feel free to tweak the code to fit your specific detection workflow.

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