python - 多处理时如何获取每个进程ID

Question

我有一些问题，因为我是 Python 和 Pyside 的新手。

我有 N 个进程同时在运行。

由于这些进程需要一些时间才能完成其工作，因此最终用户可能想要取消特定进程。所以我需要一种方法来了解进程的 ID，以便将此功能添加到程序中。

有一个answer在 Stackoverflow 中，这正是我正在做的。

代码如下:

#!/usr/bin/env python3
import multiprocessing, multiprocessing.pool, time, random, sys
from PySide.QtCore import *
from PySide.QtGui import *

def compute(num_row):
    print("worker started at %d" % num_row)
    random_number = random.randint(1, 10)
    for second in range(random_number):
        progress = float(second) / float(random_number) * 100
        compute.queue.put((num_row, progress,))
        time.sleep(1)
    compute.queue.put((num_row, 100))

def pool_init(queue):
    # see https://stackoverflow.com/a/3843313/852994
    compute.queue = queue

class MainWindow(QMainWindow):
    def __init__(self):
        QMainWindow.__init__(self)
        self.toolBar = self.addToolBar("Toolbar")
        self.toolBar.addAction(QAction('Add Task', self, triggered=self.addTask))
        self.table = QTableWidget()
        self.table.verticalHeader().hide()
        self.table.setColumnCount(2)
        self.setCentralWidget(self.table)

        # Pool of Background Processes
        self.queue = multiprocessing.Queue()
        self.pool = multiprocessing.Pool(processes=4, initializer=pool_init, initargs=(self.queue,))
        # Check for progress periodically
        self.timer = QTimer()
        self.timer.timeout.connect(self.updateProgress)
        self.timer.start(2000)

    def addTask(self):
        num_row = self.table.rowCount()
        self.pool.apply_async(func=compute, args=(num_row,))
        label = QLabel("Queued")
        bar = QProgressBar()
        bar.setValue(0)
        self.table.setRowCount(num_row + 1)
        self.table.setCellWidget(num_row, 0, label)
        self.table.setCellWidget(num_row, 1, bar)

    def updateProgress(self):
        if self.queue.empty(): return
        num_row, progress = self.queue.get() # unpack
        print("received progress of %s at %s" % (progress, num_row))
        label = self.table.cellWidget(num_row, 0)
        bar = self.table.cellWidget(num_row, 1)
        bar.setValue(progress)
        if progress == 100:
            label.setText('Finished')
        elif label.text() == 'Queued':
            label.setText('Downloading')
        self.updateProgress() # recursion

if __name__ == '__main__':
    app = QApplication(sys.argv)
    main_window = MainWindow()
    main_window.show()
    sys.exit(app.exec_())

我添加了一个“停止”按钮，我知道如何获取表格中的选定行，但我不知道如何获取选定行的进程ID以终止。

更新 1:

为了让这更容易，我可以改变

multiprocessing.Pool(processes=4, initializer=pool_init, initargs=(self.queue,)) 到

multiprocessing.Pool(processes=1, initializer=pool_init, initargs=(self.queue,))

这样所有进程都必须等到一个进程结束

现在我们有一个进程在运行，其他进程在队列中，我如何才能只获得正在运行的进程的进程 ID？

Answer 1

我编写了一个或多或少重现多处理示例的演示，并添加了中止上传的功能。它一次只能处理六个并行上传，因为这是 QNetworkAccessManager 允许的最大值。但是，可以通过简单地添加另一个 QNetworkAccessManager 来增加此限制。

我在测试演示时遇到了一个问题。似乎在某些情况下，后数据可以发送两次(例如，参见 here)。但我不知道这是一个 Qt 错误，还是我的测试设置的问题(我使用了 python 线程 httpserver)。无论如何，通过关闭 uploadProgress 处理程序中的 reply-object 很容易修复(见下文)。

from PyQt4 import QtCore, QtGui, QtNetwork

class Window(QtGui.QWidget):
    def __init__(self, address):
        QtGui.QWidget.__init__(self)
        self.address = address
        self.table = QtGui.QTableWidget(self)
        header = self.table.horizontalHeader()
        header.setStretchLastSection(True)
        header.hide()
        self.table.setColumnCount(2)
        self.button = QtGui.QPushButton('Add Upload', self)
        self.button.clicked.connect(self.handleAddUpload)
        layout = QtGui.QVBoxLayout(self)
        layout.addWidget(self.table)
        layout.addWidget(self.button)
        self.netaccess = QtNetwork.QNetworkAccessManager(self)
        self._uploaders = {}

    def handleAddUpload(self):
        stream = QtCore.QFile('files/sample.tar.bz2')
        if stream.open(QtCore.QIODevice.ReadOnly):
            data = stream.readAll()
            stream.close()
            row = self.table.rowCount()
            button = QtGui.QPushButton('Abort', self.table)
            button.clicked.connect(lambda: self.handleAbort(row))
            progress = QtGui.QProgressBar(self.table)
            progress.setRange(0, len(data))
            self.table.setRowCount(row + 1)
            self.table.setCellWidget(row, 0, button)
            self.table.setCellWidget(row, 1, progress)
            uploader = self._uploaders[row] = Uploader(row, self.netaccess)
            uploader.uploadProgress.connect(self.handleUploadProgress)
            uploader.uploadFinished.connect(self.handleUploadFinished)
            uploader.upload(data, self.address)

    def handleUploadProgress(self, key, sent, total):
        print('upload(%d): %d [%d]' % (key, sent, total))
        progress = self.table.cellWidget(key, 1)
        progress.setValue(sent)

    def handleUploadFinished(self, key):
        print('upload(%d) finished' % key)
        button = self.table.cellWidget(key, 0)
        button.setDisabled(True)
        uploader = self._uploaders.pop(key)
        uploader.deleteLater()

    def handleAbort(self, key):
        try:
            self._uploaders[key].abort()
        except (KeyError, AttributeError):
            pass

class Uploader(QtCore.QObject):
    uploadProgress = QtCore.pyqtSignal(object, int, int)
    uploadFinished = QtCore.pyqtSignal(object)

    def __init__(self, key, parent):
        QtCore.QObject.__init__(self, parent)
        self._key = key
        self._reply = None

    def abort(self):
        if self._reply is not None:
            self._reply.abort()

    def upload(self, data, url):
        if self._reply is None:
            request = QtNetwork.QNetworkRequest(QtCore.QUrl(url))
            request.setHeader(
                QtNetwork.QNetworkRequest.ContentTypeHeader,
                'application/x-www-form-urlencoded')
            self._reply = self.parent().post(request, data)
            self._reply.uploadProgress.connect(self.handleUploadProgress)
            self._reply.finished.connect(self.handleFinished)

    def handleUploadProgress(self, sent, total):
        self.uploadProgress.emit(self._key, sent, total)
        if sent >= total:
            # prevent duplicated uploads
            self._reply.close()

    def handleFinished(self):
        self._reply.deleteLater()
        self._reply = None
        self.uploadFinished.emit(self._key)

if __name__ == '__main__':

    import sys
    app = QtGui.QApplication(sys.argv)
    window = Window('http://localhost:54321/upload')
    window.setGeometry(500, 300, 500, 300)
    window.show()
    sys.exit(app.exec_())