如何在 Process() 中中断 Queue.get()

Question

当在目标函数run_event()中实现队列时，我无法停止进程。队列需要从目标函数返回值。可以在没有队列的情况下停止该过程。

from PyQt5.QtWidgets import QHBoxLayout, QApplication, QWidget, QPushButton
from PyQt5.QtCore import QThread
from multiprocessing import Process, Queue,
def run_event(params, queue):
    """Run some task that cant be stopped by polling
       returns a dataframe
    """
    out = arbritrarySolver(params) # Runs a solver that returns a dataframe
    queue.put(out)

class Worker(QObject):
    def __init__(self, params):
        super().__init__()
        self.params = params
    def stop_work(self):
        self.process.terminate()
    def start_work(self):
        self.queue = Queue()
        self.process = Process(target=run_event, args=(self.params, self.queue))
        self.process.start()
        output = self.queue.get()  # output of run_event
class Gui(QWidget):
    def __init__(self):
        super().__init__()
        self.initUI()
    def start_event(self):
        self.worker_thread = QThread()
        self.obj = worker.Worker(params)
        self.obj.moveToThread(self.worker_thread)
        self.worker_thread.started.connect(self.obj.start_work)
        self.stopButton.clicked.connect(self.obj.stop_work)
        self.worker_thread.start()
    def initUI(self):
        self.runButton = QPushButton('Run', self)
        self.runButton.clicked.connect(self.start_event)
        hbox = QHBoxLayout()
        hbox.addWidget(self.runButton)
        self.setLayout(hbox)
        self.show()
if __name__ == "__main__":
    app = QApplication(sys.argv)
    ex = Gui()
    sys.exit(app.exec_())

如何实现在

实现队列时使用停止按钮中断run_event()的解决方案？

Answer 1

问题是，当您使用 self.queue.get(( 时，您正在阻塞工作线程所在的辅助线程，相反，一种可能的策略是使用计时器来检查队列的状态，并相应地向 GUI 发送带有结果的信号。

from functools import partial
import multiprocessing as mp
from PyQt5 import QtCore, QtWidgets
from foo_package import arbritrarySolver

def run_event(params, queue):
    """Run some task that cant be stopped by polling
       returns a dataframe
    """
    out = arbritrarySolver(params)  # Runs a solver that returns a dataframe
    queue.put(out)

class Worker(QtCore.QObject):
    resultChanged = QtCore.pyqtSignal(object)
    finished = QtCore.pyqtSignal()
    def __init__(self):
        super().__init__()
        self.m_timer = QtCore.QTimer(self)
        self.m_timer.timeout.connect(self.check_result)
        self.process = None
    @QtCore.pyqtSlot()
    def stop_work(self):
        if isinstance(self.process, mp.Process):
            self.process.terminate()
    @QtCore.pyqtSlot(object)
    def start_work(self, params):
        self.queue = mp.Queue()
        self.process = mp.Process(target=run_event, args=(params, self.queue))
        self.process.start()
        self.m_timer.start(0)
    @QtCore.pyqtSlot()
    def check_result(self):
        if not self.process.is_alive():
            self.finished.emit()
            self.m_timer.stop()
        if not self.queue.empty():
            r = self.queue.get(block=False)
            self.resultChanged.emit(r)

class Gui(QtWidgets.QWidget):
    def __init__(self):
        super().__init__()
        self.worker_thread = QtCore.QThread(self)
        self.worker_thread.start()
        self.obj = Worker()
        self.obj.moveToThread(self.worker_thread)
        self.obj.resultChanged.connect(self.onResultChanged)
        self.obj.finished.connect(self.onFinished)
        self.initUI()
    @QtCore.pyqtSlot()
    def start_event(self):
        params = "aaa", "bbb"
        wrapper = partial(self.obj.start_work, params)
        QtCore.QTimer.singleShot(0, wrapper)
        self.runButton.setDisabled(True)
        self.stopButton.setDisabled(False)
    @QtCore.pyqtSlot()
    def onFinished(self):
        self.runButton.setDisabled(False)
        self.stopButton.setDisabled(True)
    @QtCore.pyqtSlot(object)
    def onResultChanged(self, result):
        print(result)
    def initUI(self):
        self.stopButton = QtWidgets.QPushButton("Stop")
        self.stopButton.setDisabled(True)
        self.runButton = QtWidgets.QPushButton("Run")
        self.runButton.clicked.connect(self.start_event)
        self.stopButton.clicked.connect(self.obj.stop_work)
        hbox = QtWidgets.QHBoxLayout(self)
        hbox.addWidget(self.runButton)
        hbox.addWidget(self.stopButton)
        self.show()
    def closeEvent(self, event):
        self.obj.stop_work()
        super().closeEvent(event)

if __name__ == "__main__":
    import sys
    app = QtWidgets.QApplication(sys.argv)
    ex = Gui()
    sys.exit(app.exec_())

Answer 2

您可以使用 timeout 参数与time.sleep结合使用multiprocessing.Queue，以定期取消阻止执行和处理事件(例如异常(：

import time
import multiprocessing as mp

def run_event(params, queue):
    time.sleep(10)
    queue.put(params)

if __name__ == '__main__':
    queue = mp.Queue()
    process = mp.Process(target=run_event, args=('foo', queue))
    process.start()
    while True:
        try:
            result = queue.get(timeout=0.1)
        except mp.queues.Empty:
            pass
        else:
            break
    print(result)

在您的示例中，您可以将while True替换为类似 while not self._stop_execution 的内容，并在单击按钮时设置self._stop_execution = True。