Python Flask多线程应用：线程控制与优雅关闭问题咨询

问题背景 #

我有三个文件（main.py、bms.py、app.py），需求是通过main.py控制Flask Web服务器和AntBMS线程，AntBMS线程要向Web服务器实时更新数据。目前遇到两个问题：

• 客户端连接Web服务器时，服务器会创建新线程
• 终止程序时，AntBMS线程无法正确join，出现KeyboardInterrupt异常

当前运行现象 #

Press CTRL+C to quit
[DEBUG][bms] Going to sleep for 5s
[DEBUG][bms] Going to sleep for 5s
^C[INFO][MainThread] Stop
stop
^CException ignored in: <module 'threading' from ...
in _shutdown
lock.acquire()
KeyboardInterrupt:

修正后代码 #

app.py（无修改） #

from flask import Flask, render_template
from flask_socketio import SocketIO

app = Flask(__name__)
socketio = SocketIO(app, async_mode=None)


@app.route('/')
def index():
    return render_template('index.html')


@socketio.on('connect', namespace='/test')
def test_connect():
    pass

bms.py（添加线程控制逻辑） #

from app import socketio
import configparser
import logging
import threading
import time

from random import random

BMS_CFG = 'bms.ini'

LOGGER = logging.getLogger(__name__)

config = configparser.ConfigParser()
config.read_dict(
    {
        'CORE': {
            'main_sleep_time': 5,
            'read_sleep_time': 5
        },
        'LOGGER': {
            'severity': logging.INFO
        },
        'WEBSERVER': {
            'host': '0.0.0.0',
            'port': 80,
            'debug': False
        }
    }
)
config.read(BMS_CFG)


def logger_init(log_level: int):
    LOGGER.setLevel(logging.DEBUG)
    stream = logging.StreamHandler()
    stream.setLevel(log_level)
    fmt = logging.Formatter('[%(levelname)s][%(threadName)s] %(message)s')
    stream.setFormatter(fmt)
    LOGGER.addHandler(stream)


logger_init(config.get('LOGGER', 'severity'))


class AntBMS():
    def __init__(self):
        LOGGER.debug("init")
        self.running = threading.Event()
        self.running.set()  # 初始化标记为运行状态

    def run(self):
        LOGGER.info("Starting updating thread")
        try:
            while self.running.is_set():
                # 生成模拟数据
                t = round(random()*10, 3)
                socketio.emit('message', {'data': t}, namespace='/test')
                sleep_time = config.getint('CORE', 'main_sleep_time')
                LOGGER.debug(f"Going to sleep for {sleep_time}s")
                # 使用Event.wait替代sleep，支持提前唤醒
                self.running.wait(sleep_time)
        except Exception as ex:
            LOGGER.error(f"Error in BMS thread: {str(ex)}")
    
    def stop(self):
        LOGGER.info("Stopping BMS thread")
        self.running.clear()  # 清除运行标记，终止循环

main.py（添加信号处理与线程join） #

from app import app, socketio
from bms import AntBMS, config, LOGGER
import threading
import signal

def handle_exit(signum, frame):
    LOGGER.info("Received exit signal, shutting down...")
    # 先停止BMS线程
    ant.stop()
    # 停止SocketIO服务器
    socketio.stop()

if __name__ == '__main__':
    print("start")

    ant = AntBMS()
    bms_thread = threading.Thread(target=ant.run, name="bms")
    bms_thread.start()

    # 注册信号处理器，捕获Ctrl+C(SIGINT)和系统终止信号(SIGTERM)
    signal.signal(signal.SIGINT, handle_exit)
    signal.signal(signal.SIGTERM, handle_exit)

    # 启动SocketIO服务器
    socketio.run(app, host="0.0.0.0", port=80, debug=False, use_reloader=False)
    
    # 等待BMS线程完全退出
    bms_thread.join()
    LOGGER.info("Application stopped")
    print("stop")

关键修改说明 #

1. 优雅终止BMS线程：

   • 在AntBMS类中加入threading.Event作为运行状态标记，替代无限循环+sleep的模式
   • 使用self.running.wait(sleep_time)替代socketio.sleep()，收到停止信号时可立即唤醒线程，无需等待sleep结束
   • 添加stop()方法，用于清除运行标记，触发线程退出

2. 处理退出信号：

   • 在main.py中注册SIGINT和SIGTERM信号处理器，确保收到退出指令时先停止BMS线程，再关闭Web服务器
   • 避免直接中断导致线程残留

3. 等待线程完成：

   • 在socketio.run()返回后调用bms_thread.join()，确保BMS线程完全退出后再结束主程序，解决KeyboardInterrupt异常

4. 关于Web服务器创建新线程：

   • Flask-SocketIO为每个客户端连接创建线程是框架的正常行为，用于处理SocketIO实时通信，属于预期机制，无需额外修改

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