终止子流程。在PyQt5中打开

debugcn 发表于 Dev

霍亚

我想取消由subprocess.Popen运行的进程。

该代码用于说明。

test_ui.py

import sys
from PyQt5.QtWidgets import (
    QApplication,
    QWidget,
    QHBoxLayout,
    QVBoxLayout,
    QPushButton,
    QTextEdit,
)
from test_worker import TestWorker


class TestUI(QWidget):
    def __init__(self):
        super().__init__()

        self.worker = TestWorker()
        self.worker.outSignal.connect(self.logging)

        self.result = QTextEdit()

        self.init_ui()

    def init_ui(self):
        start_button = QPushButton("Start")
        start_button.clicked.connect(self.start)

        cancel_button = QPushButton("Cancel")
        cancel_button.clicked.connect(self.cancel)

        hlayout = QHBoxLayout()
        hlayout.addWidget(start_button)
        hlayout.addWidget(cancel_button)

        vlayout = QVBoxLayout()
        vlayout.addLayout(hlayout)
        vlayout.addWidget(self.result)

        self.setLayout(vlayout)
        self.show()

    def start(self):
        self.result.clear()
        self.worker.run_command(["ping stackoverflow.com -n 30"])

    def cancel(self):
        self.worker.cancel_command()

    def logging(self, msg):
        msg = msg.strip()
        if msg != "":
            self.result.append(msg)


if __name__ == "__main__":
    APP = QApplication(sys.argv)
    ex = TestUI()
    sys.exit(APP.exec_())

test_worker.py

import os
import signal
import threading
import subprocess
from PyQt5.QtCore import QObject, pyqtSignal


class TestWorker(QObject):
    outSignal = pyqtSignal(str)

    def __init__(self):
        super().__init__()

    def run_command(self, cmd):
        threading.Thread(target=self._execute_command, args=(cmd), daemon=True).start()

    def cancel_command(self):
        # self.proc.kill()
        # os.kill(self.proc.pid, signal.SIGTERM)

    def _execute_command(self, cmd):
        self.proc = subprocess.Popen(
            cmd,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            shell=True,
            universal_newlines=True,
        )
        for line in self.proc.stdout:
            self.outSignal.emit(line)

self.proc.kill()=>无效。但是需要“ shell = True”。

os.kill(self.proc.pid, signal.SIGTERM)=>它也不起作用。而且，如果我按两次“取消”按钮，则会出现错误。

“ FileNotFoundError：[WinError 2]系统找不到指定的文件”

请帮助我解决这个问题。

永乐

您必须使用一个标志来指示应取消该进程，然后终止该进程并完成线程执行。Athreading.Event可以用于这种情况。

class TestWorker(QObject):
    outSignal = pyqtSignal(str)

    def __init__(self):
        super().__init__()
        self.stop_event = threading.Event()

    def run_command(self, cmd):
        self.stop_event.clear()
        threading.Thread(target=self._execute_command, args=(cmd), daemon=True).start()

    def cancel_command(self):
        self.stop_event.set()

    def _execute_command(self, cmd):
        if self.stop_event.isSet():
            return
        proc = subprocess.Popen(
            cmd,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            shell=True,
            universal_newlines=True,
        )
        if self.stop_event.isSet():
            proc.kill()
            return
        for line in proc.stdout:
            if self.stop_event.isSet():
                proc.kill()
                return
            self.outSignal.emit(line)

本文收集自互联网，转载请注明来源。

如有侵权，请联系[email protected] 删除。