如何在Python PyQt中中断QThread上的脚本执行？

Ah, I’ve been there! Trying to stop a script running in a QThread via exec() is tricky because exec() blocks the thread, and forcing a QThread termination is almost always a bad idea (PyQt explicitly warns against it for good reason—resource leaks, deadlocks, you name it). Let’s break down the proper ways to fix this:

1. Use subprocess Instead of exec() (Recommended) #

The cleanest approach is to run the script as a separate subprocess rather than executing it directly in the QThread. This lets you safely terminate the subprocess without risking your main application’s stability.

Example Thread Class #

from PyQt5.QtCore import QThread, pyqtSignal
import subprocess
import sys

class ScriptRunner(QThread):
    # Signals to communicate with main thread
    new_output = pyqtSignal(str)
    run_finished = pyqtSignal(bool)

    def __init__(self, script_path):
        super().__init__()
        self.script_path = script_path
        self._should_stop = False
        self._process = None

    def run(self):
        try:
            # Launch the script as a subprocess
            self._process = subprocess.Popen(
                [sys.executable, self.script_path],
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                text=True,
                bufsize=1,
                universal_newlines=True
            )

            # Stream output to the main thread
            for line in iter(self._process.stdout.readline, ""):
                if self._should_stop:
                    # Terminate the subprocess if stop is requested
                    self._process.terminate()
                    break
                self.new_output.emit(line.strip())

            # Wait for the process to exit and get its return code
            exit_code = self._process.wait()
            self.run_finished.emit(exit_code == 0)
        except Exception as e:
            self.new_output.emit(f"Error running script: {str(e)}")
            self.run_finished.emit(False)

    def stop(self):
        # Flag the thread to stop and terminate the subprocess
        self._should_stop = True
        if self._process and self._process.poll() is None:
            self._process.terminate()
            # If terminate doesn't work after a timeout, use kill() as a last resort
            # self._process.wait(timeout=2)
            # if self._process.poll() is None:
            #     self._process.kill()

Connect the Stop Button in Your Main Window #

# Assume you have a QPushButton named stop_btn and an instance of ScriptRunner
self.runner = ScriptRunner(path_to_your_script)
self.stop_btn.clicked.connect(self.on_stop_clicked)

def on_stop_clicked(self):
    if self.runner.isRunning():
        self.runner.stop()
        self.runner.wait()  # Wait for the thread to clean up safely

2. Add Interruption Checkpoints If You Must Use exec() #

If you can’t use subprocess (e.g., you’re executing script content directly instead of a file), you need to modify the script to periodically check for a stop signal. This requires cooperation from the script itself.

Thread Class with Stop Flag #

from PyQt5.QtCore import QThread, pyqtSignal

class InlineScriptRunner(QThread):
    new_output = pyqtSignal(str)
    run_finished = pyqtSignal(bool)

    def __init__(self, script_content):
        super().__init__()
        self.script_content = script_content
        self.stop_requested = False

    def run(self):
        try:
            # Inject a stop flag into the script's global namespace
            script_globals = {
                "check_stop": self.check_stop,
                "print": lambda msg: self.new_output.emit(msg)  # Redirect print to signal
            }

            # Execute the script
            exec(self.script_content, script_globals)
            self.run_finished.emit(True)
        except Exception as e:
            self.new_output.emit(f"Error executing script: {str(e)}")
            self.run_finished.emit(False)

    def check_stop(self):
        # Call this function periodically in the script to check for stop requests
        return self.stop_requested

    def stop(self):
        self.stop_requested = True

Modified User Script Example #

The user’s script must include calls to check_stop() at regular intervals:

# Example script content that supports interruption
import time

for i in range(100):
    print(f"Processing step {i}...")
    time.sleep(1)
    # Check if stop is requested every iteration
    if check_stop():
        print("Stopping early due to user request.")
        break

Key Notes to Remember #

• Never use QThread.terminate(): This abruptly kills the thread at an arbitrary point, which can leave resources locked, files corrupted, or your app in an unstable state.
• Subprocess is safer: It isolates the script execution from your main app, so terminating it doesn’t risk crashing your PyQt application.
• Cooperative interruption is best: If you must run code directly in a QThread, design the script to check for stop signals voluntarily.

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