子进程退出时终止Flask服务主进程的实现问题

你的核心问题是socketio.run()会阻塞主进程，导致后续的p.join()永远执行不到，无法感知子进程的退出状态，进而没法在子进程终止时关闭主进程。下面给出三种可行的解决思路：

方法一：用线程监控子进程状态 #

启动一个后台线程专门监听子进程的状态，一旦子进程退出，立即终止SocketIO服务。这种方法跨平台兼容，逻辑简单易懂。

import sys
import threading
from multiprocessing import Process
from flask import Flask
from flask_socketio import SocketIO

app = Flask(__name__)
socketio = SocketIO(app)

def listener():
    try:     
        # 硬件连接逻辑
        print("硬件连接成功")
        # 模拟子进程运行5秒后退出
        import time
        time.sleep(5)
    except Exception as e:  
        print(f"硬件未连接: {str(e)}")
        sys.exit()
    while True:
        # 读取硬件数据的逻辑
        pass

def monitor_child_process(child_proc):
    # 等待子进程终止
    child_proc.join()
    print("子进程已终止，将关闭主进程...")
    # 停止SocketIO服务
    socketio.stop()
    # 若stop未生效，直接强制退出主进程
    sys.exit(0)

if __name__ == "__main__":
    p = Process(target=listener, args=())
    p.start() 
    # 启动监控线程，设置daemon=True确保主进程退出时线程随之结束
    monitor_thread = threading.Thread(target=monitor_child_process, args=(p,), daemon=True)
    monitor_thread.start()
    # 启动SocketIO服务
    socketio.run(app, host='0.0.0.0', port=81)

方法二：利用信号处理（仅类Unix系统有效） #

在Linux/macOS下，可以通过监听SIGCHLD信号来感知子进程的退出事件，触发主进程终止逻辑。但Windows系统不支持该信号，兼容性有限。

import signal
import sys
from multiprocessing import Process
from flask import Flask
from flask_socketio import SocketIO

app = Flask(__name__)
socketio = SocketIO(app)

def listener():
    try:     
        print("硬件连接成功")
        import time
        time.sleep(5)
        sys.exit()
    except Exception as e:  
        print(f"硬件未连接: {str(e)}")
        sys.exit()
    while True:
        pass

def handle_child_exit(signum, frame):
    print("检测到子进程退出，关闭主进程")
    socketio.stop()
    sys.exit(0)

if __name__ == "__main__":
    # 注册SIGCHLD信号处理函数
    signal.signal(signal.SIGCHLD, handle_child_exit)
    p = Process(target=listener, args=())
    p.start() 
    socketio.run(app, host='0.0.0.0', port=81)

方法三：使用SocketIO后台任务 #

利用Flask-SocketIO自带的后台任务机制，定期检查子进程的存活状态，一旦发现子进程终止就关闭服务。本质和线程监控法类似，但不需要额外导入threading模块。

import sys
from multiprocessing import Process
from flask import Flask
from flask_socketio import SocketIO

app = Flask(__name__)
socketio = SocketIO(app)

def listener():
    try:     
        print("硬件连接成功")
        import time
        time.sleep(5)
        sys.exit()
    except Exception as e:  
        print(f"硬件未连接: {str(e)}")
        sys.exit()
    while True:
        pass

def monitor_child(child_proc):
    while True:
        if not child_proc.is_alive():
            print("子进程已终止，关闭服务")
            socketio.stop()
            sys.exit(0)
        # 每隔1秒检查一次子进程状态
        socketio.sleep(1)

if __name__ == "__main__":
    p = Process(target=listener, args=())
    p.start() 
    # 启动SocketIO后台监控任务
    socketio.start_background_task(target=monitor_child, args=(p,))
    socketio.run(app, host='0.0.0.0', port=81)

各方法优缺点对比 #

• 线程监控法：跨平台兼容，逻辑清晰，无额外依赖，推荐作为首选方案。
• 信号处理法：仅适用于类Unix系统，代码简洁但兼容性差。
• SocketIO后台任务法：贴合框架特性，无需额外导入线程模块，但检查间隔会存在一定延迟。

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