如何在无循环的PySide线程中实现进程的中止/暂停/继续？

针对无循环场景下的线程中止、暂停/继续需求，核心思路是在自动化流程中插入协作式检查点，让线程主动响应控制指令。以下是具体实现方案：

1. 定义线程安全的控制类 #

创建共享控制对象，传递中止、暂停信号，确保线程间通信安全：

from threading import Event, Lock

class ThreadController:
    def __init__(self):
        self._abort_event = Event()
        self._pause_event = Event()
        self._pause_event.set()  # 初始状态为运行
        self._lock = Lock()

    def request_abort(self):
        with self._lock:
            self._abort_event.set()
            self._pause_event.set()  # 唤醒暂停的线程使其退出

    def request_pause(self):
        with self._lock:
            self._pause_event.clear()

    def request_resume(self):
        with self._lock:
            self._pause_event.set()

    def should_abort(self):
        with self._lock:
            return self._abort_event.is_set()

    def wait_if_paused(self):
        # 暂停时阻塞，直到收到继续/中止信号
        self._pause_event.wait()
        return not self.should_abort()

2. 改造自动化任务类 #

在runTest的关键逻辑节点插入检查点，主动响应控制指令：

class Test(object):
    def __init__(self, controller: ThreadController, **kwargs):
        self.controller = controller
        self.somevars = None

    def runTest(self):
        try:
            # 步骤1：初始化SAP连接
            print("初始化SAP连接")
            if self.controller.should_abort():
                return
            self.controller.wait_if_paused()

            # 步骤2：执行登录操作
            print("执行SAP登录")
            if self.controller.should_abort():
                return
            self.controller.wait_if_paused()

            # ... 重复上述检查逻辑，在你的3000-5000行代码的每个关键块后插入检查点
            # 比如每个SAP操作、数据处理块完成后都要加检查

            print("自动化流程执行完成")
        finally:
            # 确保资源释放（关闭SAP连接、清理临时文件等）
            print("释放所有资源")

3. 用QThread管理线程（替代原生Thread） #

QThread更好地集成Qt信号槽，方便UI状态同步：

from PySide6.QtCore import QThread, Signal
from PySide6.QtWidgets import QPushButton

class ProcessWorker(QThread):
    finished = Signal()
    aborted = Signal()

    def __init__(self, task_class, controller):
        super().__init__()
        self.task_class = task_class
        self.controller = controller

    def run(self):
        task = self.task_class(self.controller)
        task.runTest()
        self.aborted.emit() if self.controller.should_abort() else self.finished.emit()

class PageProcess(QWidget):
    def __init__(self):
        super().__init__()
        mainObject = QHBoxLayout(self)
        mainObject.setContentsMargins(0, 0, 0, 0)
        mainObject.setSpacing(5)
        frameLeft = CustomFrame(orientation="vertical")
        self.leftData = frameLeft.layoutObject

        mainObject.addWidget(frameLeft, stretch=1.9)
        self.worker = None
        self.controller = None
    
    def add_ProcessButton(self, thisObject: object):
        self.start_btn = QPushButton(text="Start")
        self.pause_btn = QPushButton(text="Pause")
        self.pause_btn.setEnabled(False)
        
        self.leftData.addWidget(self.start_btn)
        self.leftData.addWidget(self.pause_btn)
        
        self.start_btn.clicked.connect(lambda: self.toggle_process(thisObject))
        self.pause_btn.clicked.connect(self.toggle_pause)
        return self.start_btn
    
    def toggle_process(self, thisObject: object):
        if self.start_btn.isChecked():
            self.start_btn.setText("Cancel")
            self.pause_btn.setEnabled(True)
            # 初始化控制器和工作线程
            self.controller = ThreadController()
            self.worker = ProcessWorker(thisObject, self.controller)
            self.worker.finished.connect(self.reset_ui)
            self.worker.aborted.connect(self.reset_ui)
            self.worker.start()
        else:
            if self.controller:
                self.controller.request_abort()
    
    def toggle_pause(self):
        if not self.controller:
            return
        if self.pause_btn.text() == "Pause":
            self.controller.request_pause()
            self.pause_btn.setText("Resume")
        else:
            self.controller.request_resume()
            self.pause_btn.setText("Pause")
    
    def reset_ui(self):
        self.start_btn.setText("Start")
        self.start_btn.setChecked(False)
        self.pause_btn.setText("Pause")
        self.pause_btn.setEnabled(False)
        self.worker = None
        self.controller = None

关键注意事项 #

• 检查点必须覆盖关键节点：如果你的自动化代码存在无法拆分的阻塞调用（比如第三方库的一次性耗时操作），需要优先查看该库是否提供中断接口；若没有，可考虑用multiprocessing替代线程（进程支持强制终止，但会增加资源开销）。
• 资源必须安全释放：在runTest的finally块中处理资源清理，避免中止操作导致的资源泄漏。
• 线程安全优先：所有控制变量的读写都需通过锁保护，避免多线程冲突。

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