Windows系统下Flask开发服务器无法正常关闭的问题求助

问题 #

在Windows笔记本上使用Python 3.9运行Flask默认开发服务器，需要实现服务的按需启停，但目前无论是在VS Code还是命令行运行，都只能强制终止服务，命令行甚至需要通过任务管理器结束Python进程。尝试过参考相关方案（包括werkzeug的shutdown API、发送SIGTERM信号），但都无法正常关闭服务器，尝试的代码如下：

from flask import Flask, request, g, make_response
import queue
import time
import os
from threading import Thread
import signal
import ctypes

from multiprocessing import Process

import sys

OUTPUT_EMPTY_LIMIT = 3 
DEFAULT_PORT = 9090

SHUTDOWN    = False
EXIT_CHARS  = "Qq\x11\x18"      # 'Q', 'q', Ctrl-Q, ESC


print("Name = {}".format(__name__))

app = Flask(__name__)


def shutdown_app():
    global SHUTDOWN
    SHUTDOWN = True

@app.before_request
def before_request():
    if SHUTDOWN:
        return 'Shutting down...'
    
def run_app(port, host, debug, threaded):
    app.run(port=port, host=host, debug=debug, threaded=threaded)

    
@app.route("/")
def app_help():
    return "Hello, World from help!"

@app.route("/square/<int:number>")
def square(number):
    return str(number**2)

# This doesnt kill the app as well.
@app.route("/stop")
def stop():
    func = request.environ.get('werkzeug.server.shutdown')
    if func is None:
        raise RuntimeError('Not running with the werkzeug Server')
    func()
    return 'Server shutting down...'
    exit(0)

if __name__ == "__main__":
    port = 5000
    host = '0.0.0.0'
    debug = False
    threaded = True   
    process = Process(target=run_app, args=(port, host, debug, threaded))
    process.start()
    
    while True:
        ch = input("Enter 'Q' to stop server and exit: ")
        if (ch == 'Q'):
            break
    
    shutdown_app()
    process.terminate()
    #process.kill()
    #process.join()
    print("Exiting the app")
    # os.kill(os.getpid(), signal.SIGINT)
    os._exit(0)

解决方案 #

你的代码存在两个核心问题导致无法正常关闭服务：

• 跨进程全局变量不共享：SHUTDOWN是主进程的全局变量，修改它不会影响运行Flask服务的子进程，所以before_request里的判断完全无效。
• 关闭逻辑未联动：主进程直接调用process.terminate()时，Flask服务没有机会执行正常的资源清理，且你没有触发/stop路由的逻辑。

以下是修改后的可正常运行代码：

from flask import Flask, request
from multiprocessing import Process
import requests
import time

app = Flask(__name__)

def run_app(port, host, debug, threaded):
    app.run(port=port, host=host, debug=debug, threaded=threaded)

@app.route("/")
def app_help():
    return "Hello, World from help!"

@app.route("/square/<int:number>")
def square(number):
    return str(number**2)

@app.route("/stop")
def stop():
    func = request.environ.get('werkzeug.server.shutdown')
    if func is None:
        raise RuntimeError('Not running with the werkzeug Server')
    func()
    return 'Server shutting down...'

if __name__ == "__main__":
    port = 5000
    host = '0.0.0.0'
    debug = False
    threaded = True   
    process = Process(target=run_app, args=(port, host, debug, threaded))
    process.start()
    
    print("服务已启动，输入'Q'停止服务")
    while True:
        ch = input().strip()
        if ch.upper() == 'Q':
            try:
                # 调用stop路由触发服务正常关闭
                requests.get(f"http://{host}:{port}/stop")
                time.sleep(1)  # 预留时间让服务完成清理
            except requests.exceptions.ConnectionError:
                pass  # 服务可能已提前关闭，忽略报错
            process.terminate()
            process.join()
            break
    
    print("应用已退出")

关键修改说明 #

1. 移除无效的SHUTDOWN全局变量，避免跨进程变量不共享的问题。
2. 输入Q时，先通过HTTP请求调用/stop路由，让Flask服务执行werkzeug的正常 shutdown 逻辑，完成资源清理。
3. 调用process.terminate()确保子进程完全终止，再用process.join()等待子进程结束，避免主进程提前退出导致的资源残留。
4. 增加异常处理，防止服务已关闭时请求/stop路由出现报错。

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