如何在 Python 3 和 PyQt5 中实现多核处理？
Posted
技术标签:
【中文标题】如何在 Python 3 和 PyQt5 中实现多核处理？【英文标题】：How to implement multicore processing in Python 3 and PyQt5? 【发布时间】：2015-03-27 21:56:52 【问题描述】：背景：我正在尝试在 python 3.4 PyQT5 应用程序中实现多核处理。
在我有numpy.ndarrays 帧的应用程序中，想象它像一个[n,m,t] 数组。我需要处理每个[n,m,:] 数组，使用多核会线性加快我的进程。
我玩过multiprocessing，并使用部分示例脚本一起编写了一个简单的脚本，并给了我以下想法：
简单无 GUI 代码：
import multiprocessing as mp
import numpy
aa = numpy.random.rand(4,2,3)
def random_function(x):
return x,x**3
if __name__ == '__main__':
pool = mp.Pool(processes=4)
#with apply_asynch
#results = [pool.apply_async(cube, args=(aa[:,:,x],)) for x in range(0,aa.shape[2])]
#output = [p.get() for p in results]
#test_va = numpy.asarray( output)
#with apply
results = [pool.apply(random_function, args=(aa[:,:,x],)) for x in range(0,aa.shape[2])]
test_va = numpy.asarray( results)
这很有效，并且可以完成我需要它做的事情。
问题：现在当我在PyQT5 中实现这个时，我会遇到“酸洗”问题。所以按照PyQT4here 的建议，我制作了一个简单的GUI，生成一个线程并使用多处理。结果，我将相同的 GUI 复制了 4 次，但它似乎不起作用。
PyQT5 GUI 非工作代码：
import sys, time
from PyQt5.QtCore import *
from PyQt5.QtGui import *
from PyQt5.QtWidgets import *
import multiprocessing as mp
import numpy
class WorkThread(QThread):
finished = pyqtSignal(int,object)
def __del__(self):
self.wait()
def cube(x):
return x,x**3
def run(self):
aa = numpy.random.rand(4,2,3)
pool = mp.Pool(processes=4)
results = [pool.apply_async(self.cube, args=(aa[:,:,x],)) for x in range(0,aa.shape[2])]
output = [p.get() for p in results]
test_va = numpy.asarray( output)
for i in range(5):
QThread.sleep(0.3) # artificial time delay
self.finished.emit(i,test_va)
class test_multicore(QWidget):
def __init__(self, parent=None):
QWidget.__init__(self, parent)
self.setGeometry(300, 300, 280, 600)
self.setWindowTitle('Qthreads and multicore')
self.layout = QVBoxLayout(self)
self.testButton = QPushButton("test")
self.testButton.clicked.connect(self.test)
self.listwidget = QListWidget(self)
self.layout.addWidget(self.testButton)
self.layout.addWidget(self.listwidget)
self.threadPool = []
def add(self, text,random_matrix):
""" Add item to list widget """
print ("Add: " + str(text) +str(random_matrix))
self.listwidget.addItem(str(text))
self.listwidget.sortItems()
def addBatch(self,text="text",iters=6,delay=0.3):
""" Add several items to list widget """
for i in range(iters):
time.sleep(delay) # artificial time delay
self.add(text+" "+str(i), 0)
def test(self):
self.listwidget.clear()
self.addBatch("_non_thread_entries",iters=6,delay=0.3)
self.workThread = WorkThread()
self.workThread.finished[int,object].connect(self.add)
self.workThread.start()
# run
app = QApplication(sys.argv)
test = test_multicore()
test.show()
app.exec_()
我也尝试过使用Qobject 并将其传递给moveToThread 的线程，但又遇到了同样的问题。
问题：
如何在我的 Python 3.4 PyQT5 应用程序中实现多核处理？考虑一下我将在 Windows 和 Mac 上使用 cx_freeze 进行部署。
【问题讨论】：
【参考方案1】：添加
if __name__ == '__main__':
在环境创建之前确保应用程序被创建一次。
这是 Multiprocessing pyqt5 python 3.4 的工作代码/示例。
import sys, time
from PyQt5.QtCore import *
from PyQt5.QtGui import *
from PyQt5.QtWidgets import *
import multiprocessing as mp
import numpy
class WorkThread(QThread):
finished = pyqtSignal(int,object)
def __del__(self):
self.wait()
def cube(self,x):
return x,x**3
def run(self):
aa = numpy.random.rand(4,2,3)
pool = mp.Pool(processes=4)
results = [pool.apply_async(self.cube, args=(aa[:,:,x],)) for x in range(0,aa.shape[2])]
output = [p.get() for p in results]
test_va = numpy.asarray( output)
for i in range(5):
QThread.sleep(0.3) # artificial time delay
self.finished.emit(i,test_va)
class test_multicore(QWidget):
def __init__(self, parent=None):
QWidget.__init__(self, parent)
self.setGeometry(300, 300, 280, 600)
self.setWindowTitle('Qthreads and multicore')
self.layout = QVBoxLayout(self)
self.testButton = QPushButton("test")
self.testButton.clicked.connect(self.test)
self.listwidget = QListWidget(self)
self.layout.addWidget(self.testButton)
self.layout.addWidget(self.listwidget)
self.threadPool = []
def add(self, text,random_matrix):
""" Add item to list widget """
print ("Add: " + str(text) +str(random_matrix))
self.listwidget.addItem(str(text))
self.listwidget.sortItems()
def addBatch(self,text="text",iters=6,delay=0.3):
""" Add several items to list widget """
for i in range(iters):
time.sleep(delay) # artificial time delay
self.add(text+" "+str(i), 0)
def test(self):
self.listwidget.clear()
self.addBatch("_non_thread_entries",iters=6,delay=0.3)
self.workThread = WorkThread()
self.workThread.finished[int,object].connect(self.add)
self.workThread.start()
# run
if __name__ == '__main__':
app = QApplication(sys.argv)
test = test_multicore()
test.show()
app.exec_()
使用 apply_asynch 替代：
results = [pool.apply_async(cube, args=(aa[:,:,x],)) for x in range(0,aa.shape[2])]
output = [p.get() for p in results]
test_va = numpy.asarray( output)
【讨论】：
这不起作用，因为在进程之间传输数据时会发生酸洗。具体来说，对于输出变量中的生成器对象（例如 p.get()），酸洗是不可能的。对此有何建议？ 确实不行：TypeError: cannot pickle 'WorkThread' object【参考方案2】：
现在我试图弄清楚如何在它们自己的保护伞中启动多个窗口（这就是这个问题所要问的），我发现的第一件事是不能使用 QWidget（或任何 Widget）问题）在子线程中（一个错误地被称为重复的问题），因为它们只能驻留在主线程中（至少我得到了正确的答案）。好的，那么问题就变成了我可以创建多个主线程吗？我发现我可以使用多处理来做到这一点，但这并不是那么简单，因为我能找到的所有示例都没有显示如何做到这一点。因此，如果其他人可能会为此苦苦挣扎，我想我会发布我的基本实现，以便在您希望自己实现类似的东西时为您提供一条潜在的路径。
它进一步给出了这个问题的潜在答案。现在具体来说，这个例子不仅展示了如何在不同的进程中启动两个窗口，而且这两个窗口还包含一个辅助线程——因此是多处理的一种多答案。
import sys
import time
import multiprocessing as mlti
from PyQt5.QtGui import *
from PyQt5.QtWidgets import *
from PyQt5.QtCore import *
class WorkerGUI(QRunnable):
def __init__(self, InFunc):
super(WorkerGUI, self).__init__()
self.Func = InFunc
@pyqtSlot()
def run(self):
self.Func()
class ChildWindow(QWidget):
def __init__(self, name):
QWidget.__init__(self)
self.Name = name
print('Name :',name)
self.setWindowTitle(self.Name)
if name == 'One':
self.setGeometry(100,100,250,100)
else:
self.setGeometry(100,500,250,100)
self.CountThis = False
self.StartThis = True
self.RunThis = True
self.btnCnt = QPushButton('Cnt')
self.btnCnt.clicked.connect(self.CountMe)
self.lblCntr = QLabel()
self.lblCntr.setText('0')
HBox = QHBoxLayout()
HBox.addWidget(self.btnCnt)
HBox.addWidget(QLabel(' '))
HBox.addWidget(self.lblCntr)
HBox.addStretch(1)
VBox = QVBoxLayout()
VBox.addLayout(HBox)
VBox.addStretch(1)
self.setLayout(VBox)
self.WrkrGUI = WorkerGUI(self.CountOnMe)
def CountMe(self):
if self.CountThis:
self.btnCnt.setText('Cnt')
self.CountThis = False
else:
self.btnCnt.setText('Off')
self.CountThis = True
if self.StartThis:
self.StartThis = False
self.threadpool = QThreadPool()
self.threadpool.start(self.WrkrGUI)
def CountOnMe(self):
cnt = 0
while self.RunThis:
while self.CountThis:
cnt += 1
self.lblCntr.setText(str(cnt))
time.sleep(.01)
time.sleep(.01)
def Process2():
MainThred = QApplication([])
ChildGUI = ChildWindow('Two')
ChildGUI.show()
sys.exit(MainThred.exec_())
def Process1():
MainThred = QApplication([])
ChildGUI = ChildWindow('One')
ChildGUI.show()
sys.exit(MainThred.exec_())
if __name__ == "__main__":
multiprcess1 = mlti.Process(target=Process1)
multiprcess2 = mlti.Process(target=Process2)
multiprcess1.start()
multiprcess2.start()
【讨论】：
以上是关于如何在 Python 3 和 PyQt5 中实现多核处理？的主要内容，如果未能解决你的问题，请参考以下文章
如何在 Firebase 身份验证中实现多用户帐户登录和切换？