Python3 multiprocessing.pool in PyQt5 QThread 引发 TypeError 无法腌制 'TaskThread'[QThread] 对象

Question

我有这段代码来尝试理解Python3multiprocessing.pool在PyQt5QThread中的使用：

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是一个实例方法，这意味着必须对实例进行酸洗并发送到工作进程执行。并非所有类都可以被腌制，当然也不能TaskThread(这必须以某种方式执行线程？？更一般地说，出于性能原因，您需要注意要腌制并发送到子处理的内容。

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

目前尚不清楚使用global work的最后一个示例应该做什么。global仅在函数中有意义 - 它的工作是告诉编译器在全局命名空间中分配值，而不是本地函数命名空间。它不会将变量声明为所有函数的全局变量。