PyQt5 QThread 中的Python3 multiprocessing.pool 引发 TypeError 无法pickle 'TaskThread'[QThread] 对象

Question

我有这段代码来尝试理解 Python3

multiprocessing.pool

在 PyQt5

QThread

中的使用：

import sys
from PyQt5.QtWidgets import QApplication, QMainWindow, QWidget, QPushButton, QVBoxLayout
from PyQt5.QtCore import QThread, pyqtSignal , pyqtSlot

import time

import os

import multiprocessing as mp


def work(x):
    # print('os.getpid() : {}'.format(os.getpid()), time.time(),'\n')
    time.sleep(1)
    print(x)
    
    return x
     

class TaskThread(QThread):
    
    results = pyqtSignal(object)
    
    def __init__(self, parent=None, **kwargs):
        super().__init__(parent)
        
        print("Thread app:", int(QThread.currentThread().currentThreadId()))
    
    
    @pyqtSlot()
    def run(self):
        
        print('run..................')
        
        print("Thread app run :", int(QThread.currentThread().currentThreadId())) ## i.e. 140079831918336
        
        print("Thread app run :", (QThread.currentThread())) ## i.e. <__main__.TaskThread object at 0x7f66ed6f9280>
        
        print("Thread app run :", (QThread.currentThread().currentThreadId())) ## i.e. <sip.voidptr object at 0x7f66eceb94b0>
        
        testFL = [1, 2, 3, 4, 5, 6]
        pool = mp.Pool(6)
       
        
       
        result = pool.map(self.work, testFL)  # TypeError: cannot pickle 'TaskThread' object
        
        # result = pool.map(work, testFL)  ## this works using work at top level 
        
        pool.close()
        pool.join()
        
        
        # result = []
        # for i in testFL:
            
        #     result.append(self.work(i))
            
            
        
        
        # print(result)
        # print(type(result))
        
        self.results.emit(result)
        
        print('QThread run finished')
        
    def work(self, x):
        # print('os.getpid() : {}'.format(os.getpid()), time.time())
        # time.sleep(1)
        print(x)
        return x
        
        
class MainWindow(QMainWindow):
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.setGeometry(100, 100, 300, 50)
        self.setWindowTitle('QThread Demo')
        
        # setup widget
        self.widget = QWidget()
        layout = QVBoxLayout()
        self.widget.setLayout(layout)
        self.setCentralWidget(self.widget)       

        self.btn_start = QPushButton('Start', clicked=self.start)

        layout.addWidget(self.btn_start)
        
        print("QMainWindow:", int(QThread.currentThread().currentThreadId()))
        
    def start(self):
        
        self.use_thread()


    def use_thread(self):
        self.thread = TaskThread(self)
        self.thread.finished.connect(self.task_finished)
        
        
        self.thread.results.connect(self.print_qthread_res)
        
        self.thread.start()
    
    @pyqtSlot(object)
    def print_qthread_res(self, objectz):
        
        print('passssed .......... : ', objectz)
        print('type : ', type(objectz))
    
    def task_finished(self):
        
        print('QThread FINISHED !!!!!!!!!')
        pass
    
if __name__ == '__main__':
    app = QApplication(sys.argv)
    window = MainWindow()
    window.show()
    sys.exit(app.exec())

这会产生错误：

result = pool.map(self.work, testFL)  # TypeError: cannot pickle 'TaskThread' object .... ....   TypeError: cannot pickle 'TaskThread' object

我可以通过更改来使代码工作：

result = pool.map(self.work, testFL)

至

result = pool.map(work, testFL)

使用脚本顶部定义的函数

work

而不是

 class TaskThread

函数

work

(

self.work

)。

或将

TaskThread.work

定义为：

    global work   
    
    def work( x):
        # print('os.getpid() : {}'.format(os.getpid()), time.time())
        # time.sleep(1)
        print(x)
        return x

并使用

result = pool.map(work, testFL) 

当然还有注释掉脚本顶部定义的

work

函数。

我的问题是为什么会出现错误，我的解决方案中哪一种更好？或者我应该注意其他事项以使我的脚本以正确的方式工作？

Answer 1

问题是

self.worker

是一个实例方法，这意味着实例必须被pickle并发送到工作进程执行。并不是所有的类都可以被 pickle，当然也不是

TaskThread

（它会以某种方式必须执行一个线程？？）。更一般地说，出于性能原因，您需要小心酸洗并发送到子流程的内容量。

将

work

定义为函数而不是方法很可能是最好的方法。尽可能将父类实例与子流程中正在完成的工作解耦。

尚不清楚使用

global work

的最后一个示例应该做什么。

global

仅在函数内有意义 - 它的工作是告诉编译器在全局命名空间中赋值，而不是本地函数命名空间中。它不会将变量声明为所有函数的全局变量。