如何实现软件中断？Python后端接收MQTT信号后如何立即终止线程执行

方案1：协作式软中断 （推荐，无资源泄漏风险） #

Python原生不支持跨线程强制抢占中断，最稳定的实现方式为协作式中断，只要在工作线程的线性代码中密集插入线程安全的终止标志检查，响应延迟可控制在10ms以内，完全满足普通工控场景的急停需求。
实现逻辑：

• 使用threading.Event作为全局终止标志，天生线程安全无需额外加锁
• 将原有长耗时的线性逻辑拆分为多个短耗时执行片段，每执行完一个片段就检查终止标志
• 检测到终止标志被设置时，先执行资源清理（复位GPIO、关闭IO等）再退出线程
  代码示例：

import threading
import time

# 全局线程安全终止事件
stop_event = threading.Event()
worker_thread = None

def business_logic():
    # 替换为你的原有线性条件判断/执行业务逻辑
    exec_step = 0
    while exec_step < 1000:
        # 密集插入终止检查，点位越密响应速度越快
        if stop_event.is_set():
            # 此处添加资源清理逻辑：复位硬件、关闭文件/连接等
            print("收到急停信号，线程安全退出")
            return
        # 原有业务逻辑片段
        print(f"执行步骤 {exec_step}")
        time.sleep(0.01) # 模拟单步业务耗时
        exec_step += 1
    print("业务逻辑执行完成")

# 启动工作线程接口
def start_worker():
    global worker_thread
    stop_event.clear()
    if not worker_thread or not worker_thread.is_alive():
        worker_thread = threading.Thread(target=business_logic, daemon=True)
        worker_thread.start()

# MQTT收到急停信号时调用此接口
def emergency_stop():
    stop_event.set()
    # 可选：等待线程安全退出，超时可自定义
    if worker_thread and worker_thread.is_alive():
        worker_thread.join(timeout=1)

方案2：强制终止线程 （仅适合无IO操作的纯计算场景） #

如果业务逻辑中存在无法插入终止检查的长耗时阻塞调用，可通过调用Python底层API强制终止线程，该方案存在资源泄漏风险（未释放的锁、未关闭的IO/GPIO等），仅适合特殊场景使用。
代码示例：

import ctypes
import threading

def force_terminate_thread(thread: threading.Thread):
    if not thread.is_alive():
        return
    # 向指定线程注入SystemExit异常实现强制终止
    thread_id = thread.ident
    ctypes.pythonapi.PyThreadState_SetAsyncExc(
        ctypes.c_long(thread_id),
        ctypes.py_object(SystemExit)
    )

使用方式：收到急停信号时调用该函数终止工作线程，下次启动时重新实例化Thread对象即可。 #

方案3：子进程信号终止 （最接近硬件中断效果，无需修改原有业务逻辑） #

如果需要完全模拟硬件中断的优先级抢占效果，可将业务逻辑放到独立子进程中运行，通过操作系统信号实现毫秒级的强制终止，所有资源会被操作系统自动回收，无泄漏风险。
实现逻辑：

• 业务逻辑运行在独立子进程中，和主进程（MQTT监听进程）完全隔离
• 收到急停信号时，主进程向子进程发送终止信号，操作系统会抢占子进程的CPU时间片立刻终止执行
• 下次启动时重新fork子进程即可
  代码示例：

import multiprocessing
import time

def business_logic():
    # 原有业务逻辑无需任何修改
    exec_step = 0
    while exec_step < 1000:
        print(f"执行步骤 {exec_step}")
        time.sleep(0.01)
        exec_step += 1

worker_proc = None

def start_worker_proc():
    global worker_proc
    if not worker_proc or not worker_proc.is_alive():
        worker_proc = multiprocessing.Process(target=business_logic)
        worker_proc.start()

def emergency_stop_proc():
    global worker_proc
    if worker_proc and worker_proc.is_alive():
        worker_proc.terminate() # 操作系统级抢占终止，响应速度极快
        worker_proc.join()

内容的提问来源于stack exchange，提问作者Jüri Bogatkin