试图让 QProcess 与队列一起工作

Question

我正在尝试使用队列运行多个进程并使用QProcess 获取所有进程的输出，但我遇到了几个问题。我正在使用QSpinBox 来设置同时运行的最大进程，并且我可以在主线程中让一切正常工作，或者如果我使用QObject 中的进程运行循环但我无法得到它在QThread 中正常工作。
我知道没有必要使用带有QProcess 的线程，但是对于循环，我几乎别无选择。当在主线程中运行时，它会暂时冻结，直到进程开始，我宁愿让它运行得更顺畅。
除非我使用_process.waitForFinished() 之类的东西，否则我在尝试在QThread 中运行进程时只会出错，但问题是进程一次只能运行一个。
有没有人有任何建议可以正常工作？我目前正在使用 Pyside2，但 Pyside2 或 PyQt5 的答案会很好。谢谢。

import queue
import sys
from PySide2.QtCore import QProcess, QTextCodec, QThread, Qt
from PySide2.QtWidgets import QApplication, QWidget, QSpinBox, \
    QPushButton, QVBoxLayout

class Window(QWidget):
    def __init__(self):
        QWidget.__init__(self)
        self.setAttribute(Qt.WA_DeleteOnClose, True)
        self.queue = queue.Queue()
        layout = QVBoxLayout(self)
        self.startBtn = QPushButton('Start', clicked=self.addToQueue)
        self.spinBox = QSpinBox(value=3)
        layout.addWidget(self.spinBox)
        layout.addWidget(self.startBtn)
        self.taskList = ['my.exe -value','my.exe -value','my.exe -value','my.exe -value',
                         'my.exe -value','my.exe -value','my.exe -value','my.exe -value']

    def addToQueue(self):
        for i in self.taskList:
            self.queue.put(i)
        self.sendToThread()

    def sendToThread(self):
        vals = {'max': self.spinBox.value()}
        self.taskThread = TaskThread(self.queue, vals)
        self.taskThread.start()

    def closeEvent(self, event):
        event.accept()

class TaskThread(QThread):
    def __init__(self, queue=None, vals=None, parent=None):
        QThread.__init__(self, parent)
        self.queue = queue
        self.vals = vals
        self.maxProcs = self.vals.get('max')
        self.procCount = 0

    def run(self):
        self.start_procs()

    def start_procs(self):
        while not self.queue.empty() and self.procCount < self.maxProcs:
            cmd = self.queue.get()
            _process = QProcess(self)
            _process.setProcessChannelMode(QProcess.MergedChannels)
            self.codec = QTextCodec.codecForLocale()
            self._decoder_stdout = self.codec.makeDecoder()
            _process.readyReadStandardOutput.connect(lambda process=_process: self._ready_read_standard_output(process))
            _process.started.connect(self.procStarted)
            _process.finished.connect(self.procFinished)
            _process.finished.connect(self.decreaseCount)
            _process.finished.connect(self.start_procs)
            _process.start(cmd)
            self.procCount += 1

    def _ready_read_standard_output(self, process):
        self.out = process.readAllStandardOutput()
        self.text = self._decoder_stdout.toUnicode(self.out)
        print(self.text)

    def decreaseCount(self):
        if self.procCount <= 0:
            pass
        else:
            self.procCount -= 1

    def procStarted(self):
        print('started')

    def procFinished(self):
        print('finished')

if __name__ == '__main__':
    app = QApplication(sys.argv)
    window = Window()
    window.resize(200, 100)
    window.show()
    sys.exit(app.exec_())

Answer 1

当你启动一个进程时，并不意味着它启动了，因为它可能会出现执行问题，所以在第一次启动时最好等待进程启动或启动后续进程失败，满足以下要求正在运行的进程数小于最大值或不再执行任务或尚未启动。

另一方面，我也实现了停止任务，这意味着不再添加任务，但停止之前正在执行的任务将继续执行。

如果您将最大值更改为较低的值，则在满足条件之前不会抛出更多任务。

考虑到上述情况，没有必要使用线程

import queue
from PySide2 import QtCore, QtGui, QtWidgets

class TaskManager(QtCore.QObject):
    messageChanged = QtCore.Signal(str)
    numbersTaskRunningChanged = QtCore.Signal(int)

    def __init__(self, parent=None):
        super(TaskManager, self).__init__(parent)
        self._max_task = 1
        self._queue = queue.Queue()
        self._numbers_task_running = 0
        self._running = False

    def setMaxTask(self, max_task):
        self._max_task = max_task
        if self._running:
            self.call_task()

    def maxTask(self):
        return self._max_task

    def appendTask(self, task):
        self._queue.put(task)
        self.call_task()

    def start(self):
        self._running = True
        self.call_task()

    def stop(self):
        self._running = False

    def call_task(self):
        if self._numbers_task_running < self.maxTask() and not self._queue.empty() and self._running:
            cmd = self._queue.get()
            process = QtCore.QProcess(self)
            process.setProcessChannelMode(QtCore.QProcess.MergedChannels)
            process.readyReadStandardOutput.connect(self.on_readyReadStandardOutput)
            process.finished.connect(self.on_finished)
            process.started.connect(self.on_started)
            process.errorOccurred.connect(self.on_errorOccurred)
            process.start(cmd)

    def on_readyReadStandardOutput(self):
        codec = QtCore.QTextCodec.codecForLocale()
        decoder_stdout = codec.makeDecoder()
        process = self.sender()
        text = decoder_stdout.toUnicode(process.readAllStandardOutput())
        self.messageChanged.emit(text)

    def on_errorOccurred(self, error):
        process = self.sender()
        print("error: ", error, "-", " ".join([process.program()] + process.arguments()))
        self.call_task()

    def on_finished(self):
        process = self.sender()
        self._numbers_task_running -= 1
        self.numbersTaskRunningChanged.emit(self._numbers_task_running)
        self.call_task()

    def on_started(self):
        process = self.sender()
        print("started: ", " ".join([process.program()] + process.arguments()))
        self._numbers_task_running += 1
        self.numbersTaskRunningChanged.emit(self._numbers_task_running)
        self.call_task()

class Widget(QtWidgets.QWidget):
    def __init__(self, parent=None):
        super(Widget, self).__init__(parent)
        self.setAttribute(QtCore.Qt.WA_DeleteOnClose, True)
        manager = TaskManager(self)
        task_list = # ...
        for task in task_list:
            manager.appendTask(task)

        button_start = QtWidgets.QPushButton("Start", clicked=manager.start)
        button_stop = QtWidgets.QPushButton("Stop", clicked=manager.stop)
        label = QtWidgets.QLabel("0", alignment=QtCore.Qt.AlignCenter)
        manager.numbersTaskRunningChanged.connect(label.setNum)
        spinBox = QtWidgets.QSpinBox()
        spinBox.valueChanged.connect(manager.setMaxTask)
        spinBox.setValue(3)
        textEdit = QtWidgets.QTextEdit()
        manager.messageChanged.connect(textEdit.append)

        lay = QtWidgets.QVBoxLayout(self)
        lay.addWidget(spinBox)
        lay.addWidget(button_start)
        lay.addWidget(button_stop)
        lay.addWidget(label)
        lay.addWidget(textEdit)

if __name__ == '__main__':
    import sys
    app = QtWidgets.QApplication(sys.argv)
    w = Widget()
    w.show()
    sys.exit(app.exec_())