如何在Python中串行执行shell命令且不阻塞主线程/UI

核心思路是将命令执行逻辑放到独立的工作线程中，线程内串行执行命令，通过Qt的信号槽机制和主线程通信，既不会阻塞UI，也能支持随时终止执行。

完整实现代码 #

import subprocess
from PyQt5 import QtCore, QtWidgets as qt

# 命令列表 替换为你实际的shell命令
cmd_list = [
    "echo 命令1执行中 && sleep 3",
    "echo 命令2执行中 && sleep 5",
    "echo 命令3执行中 && sleep 2"
]

class Worker(QtCore.QObject):
    # 信号：全部命令执行完成、单个命令执行结束、执行出错
    all_finished = QtCore.pyqtSignal()
    command_finished = QtCore.pyqtSignal(str, int) # 命令内容，返回码
    error_occurred = QtCore.pyqtSignal(str, str) # 命令内容，错误信息

    def __init__(self):
        super().__init__()
        self._stop_flag = False
        self._current_process = None

    @QtCore.pyqtSlot()
    def run_commands(self):
        self._stop_flag = False
        for cmd in cmd_list:
            if self._stop_flag:
                break
            try:
                # 执行命令 shell=True根据需要调整
                self._current_process = subprocess.Popen(
                    cmd,
                    shell=True,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE,
                    text=True
                )
                # 等待命令执行完成 轮询检查停止标记
                while self._current_process.poll() is None:
                    if self._stop_flag:
                        self._current_process.terminate()
                        self._current_process.wait()
                        break
                    # 避免轮询占用过多CPU
                    QtCore.QThread.msleep(100)
                # 命令执行结束 获取返回码
                return_code = self._current_process.returncode
                self.command_finished.emit(cmd, return_code)
            except Exception as e:
                self.error_occurred.emit(cmd, str(e))
        self.all_finished.emit()

    @QtCore.pyqtSlot()
    def stop(self):
        self._stop_flag = True
        if self._current_process and self._current_process.poll() is None:
            self._current_process.terminate()
            self._current_process.wait()

class Window(qt.QMainWindow):
    def __init__(self):
        super().__init__()
        central_widget = qt.QWidget()
        central_widget.setLayout(qt.QHBoxLayout())
        self.setCentralWidget(central_widget)

        # 初始化工作线程和Worker
        self.worker_thread = QtCore.QThread()
        self.worker = Worker()
        self.worker.moveToThread(self.worker_thread)

        # 绑定信号槽
        self.worker_thread.started.connect(self.worker.run_commands)
        self.worker.all_finished.connect(self.worker_thread.quit)
        # 可选：绑定命令执行回调更新UI
        self.worker.command_finished.connect(self.on_command_finished)

        # 按钮
        self.start_btn = qt.QPushButton("Start")
        self.start_btn.clicked.connect(self.run_command)
        self.stop_btn = qt.QPushButton("Stop")
        self.stop_btn.clicked.connect(self.stop_command)
        central_widget.layout().addWidget(self.start_btn)
        central_widget.layout().addWidget(self.stop_btn)

    def run_command(self):
        if not self.worker_thread.isRunning():
            self.start_btn.setEnabled(False)
            self.worker_thread.start()

    def stop_command(self):
        if self.worker_thread.isRunning():
            self.worker.stop()
            self.worker_thread.wait()
        self.start_btn.setEnabled(True)

    def on_command_finished(self, cmd, return_code):
        # 可以在这里更新UI 比如显示执行日志
        print(f"命令执行完成：{cmd}，返回码：{return_code}")
        # 所有命令执行完后恢复Start按钮
        if not self.worker_thread.isRunning():
            self.start_btn.setEnabled(True)

if __name__ == "__main__":
    import sys
    app = qt.QApplication(sys.argv)
    main_window = Window()
    main_window.show()
    sys.exit(app.exec_())

实现说明 #

• 所有命令执行逻辑都在独立的Worker对象中运行，通过moveToThread移到单独的QThread线程，完全不会占用主线程资源，UI不会卡顿
• 工作线程中按顺序遍历命令列表，前一个命令执行完成（或被终止）后才会启动下一个命令，满足串行执行的需求
• 点击Stop按钮会触发stop槽，将停止标记设为True，同时终止当前正在运行的shell进程，立即中断整个执行队列
• 可以通过扩展Worker的信号，将命令输出、执行进度等信息传递到主线程更新UI，不会出现Qt多线程更新UI的崩溃问题

注意事项 #

• 如果命令包含用户输入内容，不建议使用shell=True，避免shell注入风险，此时可以将命令按空格拆分为列表传给Popen
• 如果执行的命令会创建子进程，终止时可能需要手动杀掉整个进程组，否则可能出现子进程残留的问题

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