如何终止Flask服务器中运行的特定请求并支持ID管控？

核心思路 #

通过**请求唯一标识（UUID）+ 线程终止信号（threading.Event）**实现安全终止指定请求，同时将标识存入数据库用于后续触发终止操作，全程不影响服务器运行和其他请求处理。

实现步骤 #

1. 数据库表设计 #

先创建存储请求追踪信息的表（示例用SQLite）：

CREATE TABLE request_tasks (
    request_id TEXT PRIMARY KEY,
    thread_id INTEGER,
    create_time DATETIME DEFAULT CURRENT_TIMESTAMP,
    status TEXT DEFAULT 'running' -- 状态值：running/terminating/terminated/completed
);

2. 线程追踪与ID存储 #

在Flask请求处理流程中，为每个请求生成唯一UUID作为标识，关联终止信号并将关键信息存入数据库：

from flask import Flask, jsonify
import threading
import uuid
import sqlite3
import time

app = Flask(__name__)
# 全局字典：映射请求ID到终止信号Event，用于快速查找
request_signals = {}

def get_db_connection():
    conn = sqlite3.connect('requests.db')
    conn.row_factory = sqlite3.Row
    return conn

def long_running_task(request_id, stop_event):
    conn = get_db_connection()
    try:
        # 模拟耗时任务，每30秒检查一次终止信号
        for _ in range(120):  # 对应1小时总耗时（120*30秒）
            if stop_event.is_set():
                # 更新任务状态为已终止
                conn.execute('UPDATE request_tasks SET status = ? WHERE request_id = ?', ('terminated', request_id))
                conn.commit()
                return
            # 此处替换为实际业务逻辑
            time.sleep(30)
        # 任务正常完成
        conn.execute('UPDATE request_tasks SET status = ? WHERE request_id = ?', ('completed', request_id))
        conn.commit()
    finally:
        conn.close()
        # 任务结束后清理全局字典，释放内存
        if request_id in request_signals:
            del request_signals[request_id]

@app.route('/start-task', methods=['POST'])
def start_task():
    request_id = str(uuid.uuid4())
    stop_event = threading.Event()
    # 绑定请求ID与终止信号
    request_signals[request_id] = stop_event
    # 获取当前请求线程ID
    thread_id = threading.get_ident()
    
    # 将请求信息存入数据库
    conn = get_db_connection()
    conn.execute('INSERT INTO request_tasks (request_id, thread_id) VALUES (?, ?)', (request_id, thread_id))
    conn.commit()
    conn.close()
    
    # 启动耗时任务（Flask多线程模式下，此视图函数本身运行在独立线程中）
    long_running_task(request_id, stop_event)
    return jsonify({'request_id': request_id, 'message': 'Task started'})

@app.route('/terminate-task/<request_id>', methods=['POST'])
def terminate_task(request_id):
    conn = get_db_connection()
    # 验证请求存在且处于运行状态
    task = conn.execute('SELECT * FROM request_tasks WHERE request_id = ? AND status = ?', (request_id, 'running')).fetchone()
    if not task:
        conn.close()
        return jsonify({'error': 'Task not found or already terminated/completed'}), 404
    
    # 发送终止信号
    if request_id in request_signals:
        stop_event = request_signals[request_id]
        stop_event.set()
        # 更新状态为终止中
        conn.execute('UPDATE request_tasks SET status = ? WHERE request_id = ?', ('terminating', request_id))
        conn.commit()
        conn.close()
        return jsonify({'message': 'Termination signal sent'})
    else:
        conn.close()
        return jsonify({'error': 'Task signal not found (may have completed)'}), 404

3. 关键注意事项 #

• 禁止强制终止线程：绝对不要使用thread._stop()等危险方法，会引发资源泄漏、锁未释放等问题，必须通过任务内部定期检查终止信号实现优雅退出。
• 内存与数据维护：任务完成/终止后，及时清理全局字典中的信号条目；定期归档或删除数据库中已完成的任务记录，避免数据冗余。
• 跨重启处理：若需要服务器重启后仍能触发终止，需将终止信号状态存入数据库，重启后恢复信号映射逻辑（需额外开发适配）。

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