需要 with/as 时线程化的最佳方式（python，pyqt）
Best way to threading when with/as is needed (python, pyqt)
我正在尝试 运行 一个工人 class 使用 qthread。如果使用 workerObject = workerClass() 实例化 worker，我可以毫无问题地执行此操作。如果工作人员 class 需要始终执行 exit?
，我该怎么做
下面是尝试执行此操作的示例代码。不幸的是，通过 运行ning，exit 函数没有被执行，这很可能是因为线程不知道 with/as 语句
from PyQt4 import QtCore, QtGui
from PyQt4.QtCore import QThread, QObject, pyqtSignal, pyqtSlot
import sys
import time
try:
_fromUtf8 = QtCore.QString.fromUtf8
except AttributeError:
def _fromUtf8(s):
return s
try:
_encoding = QtGui.QApplication.UnicodeUTF8
def _translate(context, text, disambig):
return QtGui.QApplication.translate(context, text, disambig, _encoding)
except AttributeError:
def _translate(context, text, disambig):
return QtGui.QApplication.translate(context, text, disambig)
class Ui_Dialog(object):
def setupUi(self, Dialog):
Dialog.setObjectName(_fromUtf8("Dialog"))
Dialog.resize(400, 284)
self.textEdit = QtGui.QTextEdit(Dialog)
self.textEdit.setGeometry(QtCore.QRect(20, 20, 361, 191))
self.textEdit.setObjectName(_fromUtf8("textEdit"))
self.pushButton = QtGui.QPushButton(Dialog)
self.pushButton.setGeometry(QtCore.QRect(170, 250, 75, 23))
self.pushButton.setObjectName(_fromUtf8("pushButton"))
self.retranslateUi(Dialog)
QtCore.QMetaObject.connectSlotsByName(Dialog)
def retranslateUi(self, Dialog):
Dialog.setWindowTitle(_translate("Dialog", "Dialog", None))
self.pushButton.setText(_translate("Dialog", "Print", None))
class UI_MainWindow(QtGui.QMainWindow, Ui_Dialog):
def __init__(self, parent=None):
super(UI_MainWindow, self).__init__(parent)
self.setupUi(self)
self.pushButton.clicked.connect(self.run)
def run(self):
self.t = QThread()
self.x = None
with worker() as self.x:
self.x.signal.connect(self.write)
self.x.moveToThread(t)
t.started.connect(self.x.run)
t.start()
@pyqtSlot(str)
def write(self, text):
self.textEdit.append("Running")
qtObject.processEvents()
class worker(QObject):
signal = pyqtSignal(str)
def __init__(self, parent = None):
super(worker, self).__init__(parent)
def __enter__(self):
return self
def run(self):
self.signal.emit("Running")
time.wait(5)
raise Exception
def __exit__(self, exc_type, exc_value, traceback):
if exc_type:
self.signal.emit("Exception")
def main(qtObject, guiObj):
guiObj.show()
qtObject.exec_()
qtObject.processEvents()
qtObject = QtGui.QApplication(sys.argv)
guiObj = UI_MainWindow()
main(qtObject, guiObj)
LE：上下文涉及需要在 QTextEdit window 中打印来自工作人员 class 的任何失败消息。 worker class 执行期间引发的任何异常都在其退出函数中处理，因此无论发生什么错误，脚本都会优雅地结束，为用户提供有意义的信息和说明。
__exit__函数确实被执行，问题是由于线程的非阻塞性质， with 构造在工作人员 "completes" 之前立即终止（调用 worker.__exit__）并引发异常。
一种实现您想要的方法是用 try/except 语句包围 worker.run() 中的代码，以捕获任何未处理的异常并执行任何清理或发出信号。
def run(self):
try:
self.signal.emit("Running")
time.sleep(5)
raise Exception
except:
self.signal.emit("Exception")
从user3419537的解决方案开始，我最终得到了以下解决方案。第二个信号 (statusSignal) 用于将所有与异常相关的信息传回主 class，以便可以调用退出函数。
from PyQt4 import QtCore, QtGui
from PyQt4.QtCore import QThread, QObject, pyqtSignal, pyqtSlot
import sys
import time
class err(Exception):
def __init__(self, *args):
self.args = [a for a in args]
mainThread = QtCore.QThread.currentThread()
try:
_fromUtf8 = QtCore.QString.fromUtf8
except AttributeError:
def _fromUtf8(s):
return s
try:
_encoding = QtGui.QApplication.UnicodeUTF8
def _translate(context, text, disambig):
return QtGui.QApplication.translate(context, text, disambig, _encoding)
except AttributeError:
def _translate(context, text, disambig):
return QtGui.QApplication.translate(context, text, disambig)
class Ui_Dialog(object):
def setupUi(self, Dialog):
Dialog.setObjectName(_fromUtf8("Dialog"))
Dialog.resize(400, 284)
self.textEdit = QtGui.QTextEdit(Dialog)
self.textEdit.setGeometry(QtCore.QRect(20, 20, 361, 191))
self.textEdit.setObjectName(_fromUtf8("textEdit"))
self.pushButton = QtGui.QPushButton(Dialog)
self.pushButton.setGeometry(QtCore.QRect(170, 250, 75, 23))
self.pushButton.setObjectName(_fromUtf8("pushButton"))
self.retranslateUi(Dialog)
QtCore.QMetaObject.connectSlotsByName(Dialog)
def retranslateUi(self, Dialog):
Dialog.setWindowTitle(_translate("Dialog", "Dialog", None))
self.pushButton.setText(_translate("Dialog", "Print", None))
class UI_MainWindow(QtGui.QMainWindow, Ui_Dialog):
def __init__(self, parent=None):
super(UI_MainWindow, self).__init__(parent)
self.setupUi(self)
self.pushButton.clicked.connect(self.run)
def run(self):
self.t = QThread()
self.x = worker()
self.x.signal.connect(self.write)
self.x.statusSignal.connect(self.runExit)
self.x.moveToThread(self.t)
self.t.started.connect(self.x.run)
self.t.start()
@pyqtSlot(str)
def write(self, text):
self.textEdit.append(text)
qtObject.processEvents()
@pyqtSlot(object, object, object)
def runExit(self, exc_type, exc_value, exc_traceback):
self.t2 = QThread()
self.x.moveToThread(self.t2)
self.t2.started.connect(lambda: self.x.__exit__(exc_type, exc_value, exc_traceback))
self.t2.start()
class worker(QObject):
signal = pyqtSignal(str)
statusSignal = pyqtSignal(object, object, object)
def __init__(self, parent = None):
super(worker, self).__init__(parent)
def run(self):
try:
self.signal.emit("Run")
time.sleep(2)
self.moveToThread(mainThread)
raise err("err message")
except err:
exc_type, exc_value, exc_traceback = sys.exc_info()
self.statusSignal.emit(exc_type, exc_value, exc_traceback)
def __exit__(self, exc_type, exc_value, traceback):
if exc_type:
self.signal.emit(exc_value.args[0])
def main(qtObject, guiObj):
guiObj.show()
qtObject.exec_()
qtObject.processEvents()
qtObject = QtGui.QApplication(sys.argv)
guiObj = UI_MainWindow()
main(qtObject, guiObj)
很可能还需要对代码进行一些调整，但就目前而言，它可以完美运行。
我正在尝试 运行 一个工人 class 使用 qthread。如果使用 workerObject = workerClass() 实例化 worker，我可以毫无问题地执行此操作。如果工作人员 class 需要始终执行 exit?
，我该怎么做下面是尝试执行此操作的示例代码。不幸的是，通过 运行ning，exit 函数没有被执行，这很可能是因为线程不知道 with/as 语句
from PyQt4 import QtCore, QtGui
from PyQt4.QtCore import QThread, QObject, pyqtSignal, pyqtSlot
import sys
import time
try:
_fromUtf8 = QtCore.QString.fromUtf8
except AttributeError:
def _fromUtf8(s):
return s
try:
_encoding = QtGui.QApplication.UnicodeUTF8
def _translate(context, text, disambig):
return QtGui.QApplication.translate(context, text, disambig, _encoding)
except AttributeError:
def _translate(context, text, disambig):
return QtGui.QApplication.translate(context, text, disambig)
class Ui_Dialog(object):
def setupUi(self, Dialog):
Dialog.setObjectName(_fromUtf8("Dialog"))
Dialog.resize(400, 284)
self.textEdit = QtGui.QTextEdit(Dialog)
self.textEdit.setGeometry(QtCore.QRect(20, 20, 361, 191))
self.textEdit.setObjectName(_fromUtf8("textEdit"))
self.pushButton = QtGui.QPushButton(Dialog)
self.pushButton.setGeometry(QtCore.QRect(170, 250, 75, 23))
self.pushButton.setObjectName(_fromUtf8("pushButton"))
self.retranslateUi(Dialog)
QtCore.QMetaObject.connectSlotsByName(Dialog)
def retranslateUi(self, Dialog):
Dialog.setWindowTitle(_translate("Dialog", "Dialog", None))
self.pushButton.setText(_translate("Dialog", "Print", None))
class UI_MainWindow(QtGui.QMainWindow, Ui_Dialog):
def __init__(self, parent=None):
super(UI_MainWindow, self).__init__(parent)
self.setupUi(self)
self.pushButton.clicked.connect(self.run)
def run(self):
self.t = QThread()
self.x = None
with worker() as self.x:
self.x.signal.connect(self.write)
self.x.moveToThread(t)
t.started.connect(self.x.run)
t.start()
@pyqtSlot(str)
def write(self, text):
self.textEdit.append("Running")
qtObject.processEvents()
class worker(QObject):
signal = pyqtSignal(str)
def __init__(self, parent = None):
super(worker, self).__init__(parent)
def __enter__(self):
return self
def run(self):
self.signal.emit("Running")
time.wait(5)
raise Exception
def __exit__(self, exc_type, exc_value, traceback):
if exc_type:
self.signal.emit("Exception")
def main(qtObject, guiObj):
guiObj.show()
qtObject.exec_()
qtObject.processEvents()
qtObject = QtGui.QApplication(sys.argv)
guiObj = UI_MainWindow()
main(qtObject, guiObj)
LE：上下文涉及需要在 QTextEdit window 中打印来自工作人员 class 的任何失败消息。 worker class 执行期间引发的任何异常都在其退出函数中处理，因此无论发生什么错误，脚本都会优雅地结束，为用户提供有意义的信息和说明。
__exit__函数确实被执行，问题是由于线程的非阻塞性质， with 构造在工作人员 "completes" 之前立即终止（调用 worker.__exit__）并引发异常。
一种实现您想要的方法是用 try/except 语句包围 worker.run() 中的代码，以捕获任何未处理的异常并执行任何清理或发出信号。
def run(self):
try:
self.signal.emit("Running")
time.sleep(5)
raise Exception
except:
self.signal.emit("Exception")
从user3419537的解决方案开始，我最终得到了以下解决方案。第二个信号 (statusSignal) 用于将所有与异常相关的信息传回主 class，以便可以调用退出函数。
from PyQt4 import QtCore, QtGui
from PyQt4.QtCore import QThread, QObject, pyqtSignal, pyqtSlot
import sys
import time
class err(Exception):
def __init__(self, *args):
self.args = [a for a in args]
mainThread = QtCore.QThread.currentThread()
try:
_fromUtf8 = QtCore.QString.fromUtf8
except AttributeError:
def _fromUtf8(s):
return s
try:
_encoding = QtGui.QApplication.UnicodeUTF8
def _translate(context, text, disambig):
return QtGui.QApplication.translate(context, text, disambig, _encoding)
except AttributeError:
def _translate(context, text, disambig):
return QtGui.QApplication.translate(context, text, disambig)
class Ui_Dialog(object):
def setupUi(self, Dialog):
Dialog.setObjectName(_fromUtf8("Dialog"))
Dialog.resize(400, 284)
self.textEdit = QtGui.QTextEdit(Dialog)
self.textEdit.setGeometry(QtCore.QRect(20, 20, 361, 191))
self.textEdit.setObjectName(_fromUtf8("textEdit"))
self.pushButton = QtGui.QPushButton(Dialog)
self.pushButton.setGeometry(QtCore.QRect(170, 250, 75, 23))
self.pushButton.setObjectName(_fromUtf8("pushButton"))
self.retranslateUi(Dialog)
QtCore.QMetaObject.connectSlotsByName(Dialog)
def retranslateUi(self, Dialog):
Dialog.setWindowTitle(_translate("Dialog", "Dialog", None))
self.pushButton.setText(_translate("Dialog", "Print", None))
class UI_MainWindow(QtGui.QMainWindow, Ui_Dialog):
def __init__(self, parent=None):
super(UI_MainWindow, self).__init__(parent)
self.setupUi(self)
self.pushButton.clicked.connect(self.run)
def run(self):
self.t = QThread()
self.x = worker()
self.x.signal.connect(self.write)
self.x.statusSignal.connect(self.runExit)
self.x.moveToThread(self.t)
self.t.started.connect(self.x.run)
self.t.start()
@pyqtSlot(str)
def write(self, text):
self.textEdit.append(text)
qtObject.processEvents()
@pyqtSlot(object, object, object)
def runExit(self, exc_type, exc_value, exc_traceback):
self.t2 = QThread()
self.x.moveToThread(self.t2)
self.t2.started.connect(lambda: self.x.__exit__(exc_type, exc_value, exc_traceback))
self.t2.start()
class worker(QObject):
signal = pyqtSignal(str)
statusSignal = pyqtSignal(object, object, object)
def __init__(self, parent = None):
super(worker, self).__init__(parent)
def run(self):
try:
self.signal.emit("Run")
time.sleep(2)
self.moveToThread(mainThread)
raise err("err message")
except err:
exc_type, exc_value, exc_traceback = sys.exc_info()
self.statusSignal.emit(exc_type, exc_value, exc_traceback)
def __exit__(self, exc_type, exc_value, traceback):
if exc_type:
self.signal.emit(exc_value.args[0])
def main(qtObject, guiObj):
guiObj.show()
qtObject.exec_()
qtObject.processEvents()
qtObject = QtGui.QApplication(sys.argv)
guiObj = UI_MainWindow()
main(qtObject, guiObj)
很可能还需要对代码进行一些调整，但就目前而言，它可以完美运行。