如何实现带长任务与外部资源的多线程Python应用Web界面重启？

针对你的场景，核心问题在于线程无法强制终止且外部资源易泄漏，最优思路是避免在进程内部处理线程终止，转而通过进程级重启实现应用重启，同时确保外部资源被正确释放。以下是3个可落地的方案：

方案1：外部进程管理器+Flask重启触发（最低侵入性） #

不用修改现有任务逻辑，仅需给Flask加一个重启接口，配合外部脚本实现进程自动重启。

Flask代码片段 #

from flask import Flask
import os
import signal

app = Flask(__name__)

# 封装外部资源的打开/关闭逻辑
class ExternalResources:
    def __init__(self):
        self.serial_conn = None
        self.sdr_device = None

    def init(self):
        # 实际初始化串口、SDR的代码
        print("Initializing serial and SDR resources...")
        self.serial_conn = "/dev/ttyUSB0"  # 示例路径
        self.sdr_device = SoapySDR.Device()  # 示例初始化

    def cleanup(self):
        # 实际关闭资源的代码
        print("Cleaning up external resources...")
        if self.serial_conn:
            # 关闭串口逻辑
            pass
        if self.sdr_device:
            self.sdr_device.close()

resources = ExternalResources()

@app.route("/restart")
def trigger_restart():
    # 第一步：强制释放所有外部资源
    resources.cleanup()
    # 第二步：触发当前进程退出，外部脚本会自动重启
    os.kill(os.getpid(), signal.SIGTERM)
    return "Application restarting..."

if __name__ == "__main__":
    resources.init()
    app.run(host="0.0.0.0", port=5000)

外部启动脚本（start_app.sh） #

#!/bin/bash
while true; do
    echo "Starting application..."
    python3 app.py
    echo "Application exited, restarting in 2s..."
    sleep 2
done

使用方式：用chmod +x start_app.sh赋予权限后运行脚本，访问/restart即可触发重启。

方案2：将长任务/外部资源操作隔离到子进程 #

把涉及串口、SDR的长任务放到独立子进程中，主进程（Flask）仅负责调度和重启触发。进程可以强制终止，避免线程无法终止的问题。

代码片段 #

from flask import Flask
import multiprocessing
import signal
import os

app = Flask(__name__)

# 长任务函数（运行在子进程）
def long_task(exit_event):
    # 子进程独立初始化外部资源
    sdr = SoapySDR.Device()
    serial = open("/dev/ttyUSB0", "r+")
    print(f"Task started in PID {os.getpid()}")

    while not exit_event.is_set():
        # 模拟10分钟以上的长任务，定期检查退出信号
        # 实际业务逻辑写在这里
        pass

    # 退出前清理资源
    serial.close()
    sdr.close()
    print(f"Task in PID {os.getpid()} exited cleanly")

# 全局保存子进程和退出信号
task_process = None
exit_event = multiprocessing.Event()

@app.route("/start-task")
def start_task():
    global task_process, exit_event
    if task_process and task_process.is_alive():
        return "Task already running"
    exit_event.clear()
    task_process = multiprocessing.Process(target=long_task, args=(exit_event,))
    task_process.start()
    return "Task started"

@app.route("/restart")
def restart_app():
    global task_process, exit_event
    # 终止子进程
    if task_process and task_process.is_alive():
        exit_event.set()
        task_process.join(timeout=15)  # 等待15秒，超时强制杀死
        if task_process.is_alive():
            task_process.terminate()
            task_process.join()
    # 清理主进程资源后退出
    os.kill(os.getpid(), signal.SIGTERM)
    return "Restarting application..."

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

方案3：线程优雅退出（仅适用于可修改任务逻辑的场景） #

给每个线程添加退出信号，要求长任务定期检查信号并主动退出，同时释放资源。注意：如果任务逻辑包含无法修改的第三方代码，此方案无效。

代码片段 #

from flask import Flask
import threading
import time

app = Flask(__name__)

# 线程安全的退出信号字典
thread_exit_signals = {}

def long_task(task_id):
    # 初始化外部资源
    serial = open("/dev/ttyUSB0", "r+")
    print(f"Thread {task_id} started")

    while not thread_exit_signals[task_id].is_set():
        # 模拟长任务，每5秒检查一次退出信号
        time.sleep(5)
        # 实际业务逻辑

    # 退出前清理资源
    serial.close()
    print(f"Thread {task_id} exited cleanly")

@app.route("/start-task/<task_id>")
def start_task(task_id):
    if task_id in thread_exit_signals and not thread_exit_signals[task_id].is_set():
        return f"Task {task_id} already running"
    thread_exit_signals[task_id] = threading.Event()
    thread = threading.Thread(target=long_task, args=(task_id,))
    thread.daemon = False  # 禁止设为守护线程，确保资源能被清理
    thread.start()
    return f"Task {task_id} started"

@app.route("/restart")
def restart_app():
    # 通知所有线程退出
    for sig in thread_exit_signals.values():
        sig.set()
    # 等待线程清理资源（可根据任务时长调整超时）
    time.sleep(20)
    # 触发进程退出，外部脚本重启
    os.kill(os.getpid(), signal.SIGTERM)
    return "Restarting application..."

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

方案选择建议 #

1. 优先选方案1：侵入性最低，不用修改现有任务逻辑，适合快速解决问题。
2. 如果需要任务与主进程通信，选方案2：进程隔离更安全，避免任务崩溃影响主进程。
3. 仅当所有长任务逻辑都可修改时，才考虑方案3。

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