Flask服务器重载时如何等待当前运行线程结束后再重启

Flask自带的debug模式重载依赖werkzeug内置的reloader模块，默认触发重启时会直接发送终止信号结束主进程，不会等待运行中线程，可通过以下方案实现优雅重载：

开发环境debug模式适配 #

通过自定义信号监听，捕获重载触发的终止信号后，先等待所有业务线程结束再退出进程，示例代码如下：

from flask import Flask
import threading
import time
import signal
import sys

app = Flask(__name__)

# 若使用线程池可直接替换为 ThreadPoolExecutor，无需手动维护线程列表
active_threads = []
shutdown_flag = False

# 自定义业务线程类，自动追踪活跃状态
class BusinessThread(threading.Thread):
    def run(self):
        active_threads.append(self)
        try:
            super().run()
        finally:
            active_threads.remove(self)

# 模拟耗时业务接口
@app.route('/long-task')
def long_task():
    def task():
        for i in range(10):
            if shutdown_flag:
                break
            print(f"耗时任务执行中：{i+1}/10")
            time.sleep(1)
        print("任务执行完成")
    t = BusinessThread(target=task)
    t.start()
    return "任务已启动"

# 重载信号处理逻辑
def graceful_shutdown(signum, frame):
    global shutdown_flag
    shutdown_flag = True
    print("检测到代码更新，等待所有运行中线程结束...")
    # 等待所有线程结束，可添加timeout参数避免僵死线程卡住重载
    for t in active_threads:
        t.join(timeout=30)
    print("所有线程已退出，开始重载")
    sys.exit(0)

# 注册SIGINT信号监听（werkzeug重载默认发送该信号）
signal.signal(signal.SIGINT, graceful_shutdown)

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

注意：如果业务使用concurrent.futures.ThreadPoolExecutor管理线程，直接调用executor.shutdown(wait=True)即可，无需手动维护线程列表。

生产环境替代方案 #

debug模式本身存在代码执行漏洞风险，绝对不可以用于生产环境，生产环境推荐使用Gunicorn、uWSGI等成熟WSGI服务器，自带优雅重载能力，无需修改业务代码：

• Gunicorn启动时添加参数 --graceful-timeout 30，重载时会最多等待30秒让现有请求/线程处理完成再终止旧进程
• uWSGI添加配置项 reload-mercy = 30 即可实现相同效果
• 若为异步后台任务，推荐使用Celery等独立任务队列管理，服务重启不会影响队列中未执行的任务

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