如何让Python的Dash服务器定时停止或手动触发停止？

方案1：运行指定时间后自动停止 #

由于app.run_server()是阻塞调用，后续代码无法执行，我们可以通过线程实现定时停止逻辑，以下是两种适配场景的实现方式：

方式1：优雅关闭服务器（推荐关闭debug模式） #

通过调用Flask服务器的shutdown()方法停止服务，适合不需要debug模式的场景：

import threading
import time
from dash import Dash, html, dcc

app = Dash(__name__)

app.layout = html.Div([
    dcc.Textarea(
        id='textarea-example',
        value='Lorem ipsum test working example',
        style={'width': '100%', 'height': 300},
    ),
    html.Div(id='textarea-example-output', style={'whiteSpace': 'pre-line'})
])

def stop_server_after_delay(server, delay_seconds):
    time.sleep(delay_seconds)
    print(f"\n服务器运行{delay_seconds}秒后自动停止...")
    server.shutdown()

if __name__ == '__main__':
    # 获取Flask服务器实例
    server = app.server
    # 启动定时停止线程（设置为60秒，可自行修改时长）
    stop_thread = threading.Thread(target=stop_server_after_delay, args=(server, 60))
    stop_thread.daemon = True  # 主线程退出时自动结束子线程
    stop_thread.start()
    
    print("服务器启动，将在60秒后自动停止")
    app.run_server(debug=False)  # debug=True会触发进程重载，导致定时逻辑失效
    print("服务器已停止")

方式2：强制终止进程（兼容debug模式） #

如果必须启用debug=True，可以直接通过信号终止当前进程，这种方式更粗暴但能覆盖debug场景：

import os
import signal
import threading
import time
from dash import Dash, html, dcc

app = Dash(__name__)

app.layout = html.Div([
    dcc.Textarea(
        id='textarea-example',
        value='Lorem ipsum test working example',
        style={'width': '100%', 'height': 300},
    ),
    html.Div(id='textarea-example-output', style={'whiteSpace': 'pre-line'})
])

def stop_server_after_delay(delay_seconds):
    time.sleep(delay_seconds)
    print(f"\n服务器运行{delay_seconds}秒后自动停止...")
    os.kill(os.getpid(), signal.SIGTERM)

if __name__ == '__main__':
    # 启动定时停止线程（设置为60秒）
    stop_thread = threading.Thread(target=stop_server_after_delay, args=(60,))
    stop_thread.daemon = True
    stop_thread.start()
    
    print("服务器启动，将在60秒后自动停止")
    app.run_server(debug=True)

方案2：控制台输入"yes"触发停止 #

通过线程监听控制台输入，当用户输入yes时触发服务器停止：

优雅关闭版本（关闭debug模式） #

import threading
from dash import Dash, html, dcc

app = Dash(__name__)

app.layout = html.Div([
    dcc.Textarea(
        id='textarea-example',
        value='Lorem ipsum test working example',
        style={'width': '100%', 'height': 300},
    ),
    html.Div(id='textarea-example-output', style={'whiteSpace': 'pre-line'})
])

def wait_for_stop_command(server):
    while True:
        user_input = input("输入'yes'可停止服务器：\n")
        if user_input.strip().lower() == 'yes':
            print("收到停止命令，正在关闭服务器...")
            server.shutdown()
            break

if __name__ == '__main__':
    server = app.server
    # 启动输入监听线程
    input_thread = threading.Thread(target=wait_for_stop_command, args=(server,))
    input_thread.daemon = True
    input_thread.start()
    
    print("服务器启动，输入'yes'可停止服务器")
    app.run_server(debug=False)
    print("服务器已停止")

兼容debug模式版本 #

import os
import signal
import threading
from dash import Dash, html, dcc

app = Dash(__name__)

app.layout = html.Div([
    dcc.Textarea(
        id='textarea-example',
        value='Lorem ipsum test working example',
        style={'width': '100%', 'height': 300},
    ),
    html.Div(id='textarea-example-output', style={'whiteSpace': 'pre-line'})
])

def wait_for_stop_command():
    while True:
        user_input = input("输入'yes'可停止服务器：\n")
        if user_input.strip().lower() == 'yes':
            print("收到停止命令，正在关闭服务器...")
            os.kill(os.getpid(), signal.SIGTERM)
            break

if __name__ == '__main__':
    # 启动输入监听线程
    input_thread = threading.Thread(target=wait_for_stop_command)
    input_thread.daemon = True
    input_thread.start()
    
    print("服务器启动，输入'yes'可停止服务器")
    app.run_server(debug=True)

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