在不影响UI的情况下并发运行后台任务；
在构建Python GUI应用程序时，一个常见的问题是在试图执行长时间运行的后台任务时“锁定”接口。在本教程中，我将介绍在PyQt6中实现并发执行的最简单的方法之一。
基于Qt的应用程序(像大多数GUI应用程序一样)是基于事件的。这意味着执行是根据用户交互、信号和计时器来驱动的。在事件驱动的应用程序中，单击按钮会创建一个事件，应用程序随后会处理该事件以产生一些预期的输出。事件被推入和从事件队列中取出，并按顺序处理。
app = QApplication([])
window = MainWindow()
app.exec_()事件循环通过在QApplication对象上调用.exec_()开始，并在与Python代码相同的线程中运行。运行此事件循环的线程(通常称为GUI线程)还处理与主机操作系统的所有窗口通信。
默认情况下，由事件循环触发的任何执行也将在此线程中同步运行。在实践中，这意味着任何时候你的PyQt应用程序在你的代码中做一些事情，窗口通信和GUI交互都是冻结的。
如果您正在做的事情很简单，并且快速地将控制返回到GUI循环，那么用户将察觉不到这种冻结。但是，如果您需要执行较长时间运行的任务，例如打开/写入一个大文件，下载一些数据，或呈现一些复杂的图像，就会出现问题。对于您的用户，应用程序将显示为无响应(因为它是)。因为你的应用程序不再与操作系统通信，如果你点击你的应用程序，你会看到旋转的死亡之轮。没有人希望这样。
解决方案很简单：将工作从GUI线程移出(放到另一个线程中)。PyQt(通过Qt)提供了一个简单的接口来实现这一点。
为了演示多线程执行，我们需要一个应用程序。下面是一个简单的PyQt应用程序，它将允许我们演示多线程，并看到实际的结果。只需复制并粘贴到一个新文件中，并使用适当的文件名(如多线程.py)保存它。将其余代码将被添加到这个文件中。
from PyQt6.QtGui import *
from PyQt6.QtWidgets import *
from PyQt6.QtCore import *
import time
class MainWindow(QMainWindow):
def __init__(self, *args, **kwargs):
super(MainWindow, self).__init__(*args, **kwargs)
self.counter = 0
layout = QVBoxLayout()
self.l = QLabel("Start")
b = QPushButton("DANGER!")
b.pressed.connect(self.oh_no)
layout.addWidget(self.l)
layout.addWidget(b)
w = QWidget()
w.setLayout(layout)
self.setCentralWidget(w)
self.show()
self.timer = QTimer()
self.timer.setInterval(1000)
self.timer.timeout.connect(self.recurring_timer)
self.timer.start()
def oh_no(self):
time.sleep(5)
def recurring_timer(self):
self.counter +=1
self.l.setText("Counter: %d" % self.counter)
app = QApplication([])
window = MainWindow()
app.exec()运行：您应该看到一个演示窗口，其中有一个数字在向上计数。这是由一个简单的循环时间产生的，每秒发射一次。这可以看作是我们的事件循环指示器，它是一种简单的方法，可以让我们知道我们的应用程序正在正常运行。还有一个按钮上写着“DANGER!”按下它。
您将注意到，每当您按下按钮时，计数器停止跳动，应用程序完全冻结5S。在Windows上，你可能会看到窗口变白，表明它没有响应，而在Mac上，你会看到旋转的死亡之轮。
显示冻结的原因是Qt事件循环被阻止处理(和响应)窗口事件。你在窗口上的点击仍然被主机操作系统注册并发送给你的应用程序，但因为它位于你的time.sleep中，它不能接受或对它们做出反应。它们必须等到您的代码将控制权传递回Qt。
解决这个问题的最简单、也许也是最合乎逻辑的方法是在代码中接受事件。这允许Qt继续响应主机操作系统，并且您的应用程序将保持响应。你可以通过在QApplication类上使用静态的.processEvents()函数轻松做到这一点。只需在长时间运行的代码块中添加如下一行:
QApplication.processEvents()例如，长时间运行代码。我们可以把它分解成5x个1秒的睡眠，并在中间插入.processEvents。
def oh_no(self):
for n in range(5):
QApplication.processEvents()
time.sleep(1)现在，当您按下按钮时，进入休眠。但是，现在QApplication.processEvents()会间歇性地将控制传递回Qt，并允许它像正常一样响应事件。Qt将接受事件并在返回运行其余代码之前处理它们。
这是可行的，但出于几个原因，它仍然是垃圾代码。
首先，当您将控制权传递回Qt时，您的代码将不再运行。这意味着无论你试图做什么长时间运行的事情都会花费更长的时间。这绝对不是你想要的。
其次，在主事件循环(app.exec())之外处理事件，会导致你的应用程序在循环内分支到处理代码(例如，用于触发槽或事件)。如果您的代码依赖于/响应外部状态，这可能会导致未定义的行为。下面的代码演示了这一点:
from PyQt6.QtGui import *
from PyQt6.QtWidgets import *
from PyQt6.QtCore import *
import time
class MainWindow(QMainWindow):
def __init__(self, *args, **kwargs):
super(MainWindow, self).__init__(*args, **kwargs)
self.counter = 0
layout = QVBoxLayout()
self.l = QLabel("Start")
b = QPushButton("DANGER!")
b.pressed.connect(self.oh_no)
c = QPushButton("?")
c.pressed.connect(self.change_message)
layout.addWidget(self.l)
layout.addWidget(b)
layout.addWidget(c)
w = QWidget()
w.setLayout(layout)
self.setCentralWidget(w)
self.show()
def change_message(self):
self.message = "OH NO"
def oh_no(self):
self.message = "Pressed"
for n in range(100):
time.sleep(0.1)
self.l.setText(self.message)
QApplication.processEvents()
app = QApplication([])
window = MainWindow()
app.exec_()如果运行这段代码，您将看到与以前一样的计数器。按下“DANGER!”将显示的文本更改为“Pressed”，如oh_no函数的入口点所定义的那样。但是，如果在oh_no仍在运行时按下“?”按钮，您将看到消息发生了变化。状态从循环外部被改变。
这是一个简单的例子。但是，如果您的应用程序中有多个长时间运行的进程，并且每个进程都调用QApplication.processEvents()来保持运行，那么您的应用程序行为可能是不可预测的。
如果你退一步思考你想在你的应用程序中发生什么，它可能可以总结为“一些事情与其他事情同时发生”。
在PyQt应用程序中运行独立任务有两种主要方法：线程和进程。
线程共享相同的内存空间，因此可以快速启动并消耗最少的资源。共享内存使得在线程之间传递数据变得很简单，但是从不同线程读取/写入内存可能导致竞争条件或段错误。在Python GUI中，还有一个额外的问题，即多个线程被同一个全局解释器锁(GIL)绑定——这意味着非GIL释放的Python代码一次只能在一个线程中执行。然而，这并不是PyQt的主要问题，因为大部分时间都是在Python之外度过的。
进程使用独立的内存空间(以及完全独立的Python解释器)。这避免了GIL的任何潜在问题，但代价是启动时间较慢，内存开销更大，发送/接收数据更复杂。
为了简单起见，通常使用线程是有意义的，除非您有很好的理由使用进程。Qt中的子进程更适合于运行和与外部程序通信。
Qt为在其他线程中运行作业提供了一个非常简单的接口，这在PyQt中很好地公开了。这是围绕两个类构建的:QRunnable和QThreadPool。前者是您想要执行的工作的容器，而后者是将工作传递给线程的方法。
使用QThreadPool的好处是它可以为您处理工作线程的排队和执行。除了排队作业和检索结果之外，根本没有太多要做的事情。
要定义一个自定义QRunnable，您可以子类化基本QRunnable类，然后将您希望执行的代码放置在run()方法中。下面是我们长时间运行的实现。sleep job作为QRunnable。将以下代码添加到multithread.py中，位于MainWindow类定义的上方。
class Worker(QRunnable):
'''
Worker thread
'''
@pyqtSlot()
def run(self):
'''
Your code goes in this function
'''
print("Thread start")
time.sleep(5)
print("Thread complete")在另一个线程中执行我们的函数只是简单地创建一个Worker实例，然后将它传递给我们的QThreadPool实例，它将自动执行。
接下来在__init__块中添加以下内容，以设置线程池。
self.threadpool = QThreadPool()
print("Multithreading with maximum %d threads" % self.threadpool.maxThreadCount())最后，将以下代码行添加到oh_no函数中。
def oh_no(self):
worker = Worker()
self.threadpool.start(worker)现在，单击按钮将创建一个工作线程来处理(长时间运行的)任务，并通过线程池将其转到另一个线程。如果没有足够的线程来处理传入的worker，它们将被排队并在稍后按顺序执行。
尝试一下，您将看到您的应用程序现在可以毫无问题地处理点击按钮。
检查一下如果你多次按下按钮会发生什么。您应该看到您的线程立即执行，直到.maxThreadCount报告的数量。如果在已经有这个数量的活动worker之后再次按下按钮，那么后续的worker将排队，直到有一个线程可用为止。
如果你想将自定义数据传递给执行函数，你可以通过init，然后从run槽函数中通过self访问数据。
class Worker(QRunnable):
'''
Worker thread
:param args: Arguments to make available to the run code
:param kwargs: Keywords arguments to make available to the run code
'''
def __init__(self, *args, **kwargs):
super(Worker, self).__init__()
self.args = args
self.kwargs = kwargs
@pyqtSlot()
def run(self):
'''
Initialise the runner function with passed self.args, self.kwargs.
'''
print(args, kwargs)事实上，我们可以利用Python中函数是对象这一事实，并传递函数来执行，而不是每次都子类化。在接下来的构造中，我们只需要一个Worker类来处理所有的执行作业。
class Worker(QRunnable):
'''
Worker thread
Inherits from QRunnable to handler worker thread setup, signals and wrap-up.
:param callback: The function callback to run on this worker thread. Supplied args and
kwargs will be passed through to the runner.
:type callback: function
:param args: Arguments to pass to the callback function
:param kwargs: Keywords to pass to the callback function
'''
def __init__(self, fn, *args, **kwargs):
super(Worker, self).__init__()
# Store constructor arguments (re-used for processing)
self.fn = fn
self.args = args
self.kwargs = kwargs
@pyqtSlot()
def run(self):
'''
Initialise the runner function with passed args, kwargs.
'''
self.fn(*self.args, **self.kwargs)您现在可以传入任何Python函数，并让它在单独的线程中执行。
def execute_this_fn(self):
print("Hello!")
def oh_no(self):
args = (2,3)
kwargs = {'test':1, "test2":2}
worker = Worker(self.execute_this_fn, args, kwargs)
worker = Worker(self.execute_this_fn) # Any other args, kwargs are passed to the run function
# Execute
self.threadpool.start(worker)有时，能够从正在运行的worker传回状态和数据是很有帮助的。这可能包括计算的结果、引发的异常或正在进行的进展(想想进度条)。Qt提供了信号和插槽框架，它允许你这样做，并且是线程安全的，允许从运行线程直接到GUI前端的安全通信。信号允许您使用.emit值，然后在代码的其他地方由与.connect链接的slot函数拾取这些值。
下面是一个简单的WorkerSignals类，定义为包含许多示例信号。
自定义信号只能在从QObject派生的对象上定义。由于QRunnable不是从QObject派生的，我们不能直接定义那里的信号。使用自定义QObject保存信号是最简单的解决方案。
import traceback, sys
class WorkerSignals(QObject):
'''
Defines the signals available from a running worker thread.
Supported signals are:
finished
No data
error
tuple (exctype, value, traceback.format_exc() )
result
object data returned from processing, anything
'''
finished = pyqtSignal()
error = pyqtSignal(tuple)
result = pyqtSignal(object)在这个例子中，我们定义了3个自定义信号:
您可能不需要所有这些信号，但它们被包括在内，以指示可能发生的事情。在下面的代码中，我们将实现一个长时间运行的任务，该任务利用这些信号向用户提供有用的信息。
class Worker(QRunnable):
'''
Worker thread
Inherits from QRunnable to handler worker thread setup, signals and wrap-up.
:param callback: The function callback to run on this worker thread. Supplied args and
kwargs will be passed through to the runner.
:type callback: function
:param args: Arguments to pass to the callback function
:param kwargs: Keywords to pass to the callback function
'''
def __init__(self, fn, *args, **kwargs):
super(Worker, self).__init__()
# Store constructor arguments (re-used for processing)
self.fn = fn
self.args = args
self.kwargs = kwargs
self.signals = WorkerSignals()
@pyqtSlot()
def run(self):
'''
Initialise the runner function with passed args, kwargs.
'''
# Retrieve args/kwargs here; and fire processing using them
try:
result = self.fn(
*self.args, **self.kwargs
)
except:
traceback.print_exc()
exctype, value = sys.exc_info()[:2]
self.signals.error.emit((exctype, value, traceback.format_exc()))
else:
self.signals.result.emit(result) # Return the result of the processing
finally:
self.signals.finished.emit() # Done您可以将自己的处理函数连接到这些信号，以接收线程完成(或结果)的通知。
def execute_this_fn(self):
for n in range(0, 5):
time.sleep(1)
return "Done."
def print_output(self, s):
print(s)
def thread_complete(self):
print("THREAD COMPLETE!")
def oh_no(self):
# Pass the function to execute
worker = Worker(self.execute_this_fn) # Any other args, kwargs are passed to the run function
worker.signals.result.connect(self.print_output)
worker.signals.finished.connect(self.thread_complete)
# Execute
self.threadpool.start(worker)您还经常希望从长时间运行的线程接收状态信息。这可以通过传递回调来实现，您运行的代码可以向回调发送信息。这里有两个选择:定义新的信号(允许使用事件循环执行处理)或使用标准的Python函数。
在这两种情况下，您都需要将这些回调传递给目标函数才能使用它们。在下面的完整代码中使用了基于信号的方法，其中我们传递了一个int作为线程进度百分比的指示器
下面给出了一个完整的工作示例，展示了自定义QRunnable工作者以及工作者和进度信号。您应该能够轻松地将此代码适应您所开发的任何多线程应用程序。
from PyQt6.QtGui import *
from PyQt6.QtWidgets import *
from PyQt6.QtCore import *
import time
import traceback, sys
class WorkerSignals(QObject):
'''
Defines the signals available from a running worker thread.
Supported signals are:
finished
No data
error
tuple (exctype, value, traceback.format_exc() )
result
object data returned from processing, anything
progress
int indicating % progress
'''
finished = pyqtSignal()
error = pyqtSignal(tuple)
result = pyqtSignal(object)
progress = pyqtSignal(int)
class Worker(QRunnable):
'''
Worker thread
Inherits from QRunnable to handler worker thread setup, signals and wrap-up.
:param callback: The function callback to run on this worker thread. Supplied args and
kwargs will be passed through to the runner.
:type callback: function
:param args: Arguments to pass to the callback function
:param kwargs: Keywords to pass to the callback function
'''
def __init__(self, fn, *args, **kwargs):
super(Worker, self).__init__()
# Store constructor arguments (re-used for processing)
self.fn = fn
self.args = args
self.kwargs = kwargs
self.signals = WorkerSignals()
# Add the callback to our kwargs
self.kwargs['progress_callback'] = self.signals.progress
@pyqtSlot()
def run(self):
'''
Initialise the runner function with passed args, kwargs.
'''
# Retrieve args/kwargs here; and fire processing using them
try:
result = self.fn(*self.args, **self.kwargs)
except:
traceback.print_exc()
exctype, value = sys.exc_info()[:2]
self.signals.error.emit((exctype, value, traceback.format_exc()))
else:
self.signals.result.emit(result) # Return the result of the processing
finally:
self.signals.finished.emit() # Done
class MainWindow(QMainWindow):
def __init__(self, *args, **kwargs):
super(MainWindow, self).__init__(*args, **kwargs)
self.counter = 0
layout = QVBoxLayout()
self.l = QLabel("Start")
b = QPushButton("DANGER!")
b.pressed.connect(self.oh_no)
layout.addWidget(self.l)
layout.addWidget(b)
w = QWidget()
w.setLayout(layout)
self.setCentralWidget(w)
self.show()
self.threadpool = QThreadPool()
print("Multithreading with maximum %d threads" % self.threadpool.maxThreadCount())
self.timer = QTimer()
self.timer.setInterval(1000)
self.timer.timeout.connect(self.recurring_timer)
self.timer.start()
def progress_fn(self, n):
print("%d%% done" % n)
def execute_this_fn(self, progress_callback):
for n in range(0, 5):
time.sleep(1)
progress_callback.emit(n*100/4)
return "Done."
def print_output(self, s):
print(s)
def thread_complete(self):
print("THREAD COMPLETE!")
def oh_no(self):
# Pass the function to execute
worker = Worker(self.execute_this_fn) # Any other args, kwargs are passed to the run function
worker.signals.result.connect(self.print_output)
worker.signals.finished.connect(self.thread_complete)
worker.signals.progress.connect(self.progress_fn)
# Execute
self.threadpool.start(worker)
def recurring_timer(self):
self.counter +=1
self.l.setText("Counter: %d" % self.counter)
app = QApplication([])
window = MainWindow()
app.exec_()您可能已经发现了这个总体计划中的轻微缺陷——我们仍然在使用事件循环(和GUI线程)来处理我们的worker的输出。
当我们只是跟踪进度、完成或返回元数据时，这不是问题。然而，如果你有返回大量数据的worker——例如加载大文件，执行复杂的分析和需要(大)结果，或者查询数据库——通过GUI线程传递这些数据可能会导致性能问题，最好避免。
类似地，如果应用程序使用大量线程和Python结果处理程序，则可能会遇到GIL的限制。如前所述，在使用线程时，Python的执行一次仅限于单个线程。处理线程信号的Python代码可能会被worker阻塞，反之亦然。因为阻塞插槽函数会阻塞事件循环，这会直接影响GUI的响应能力。
在这些情况下，最好使用纯python线程池(例如并发futures)来将处理和线程事件处理与GUI进一步隔离。但是，请注意，任何Python GUI代码都可以阻塞其他Python代码，除非它位于单独的进程中。
页面更新：2024-03-27
本站资料均由网友自行发布提供，仅用于学习交流。如有版权问题，请与我联系，QQ：4156828
© CopyRight 2008-2024 All Rights Reserved. Powered By bs178.com 闽ICP备11008920号-3
闽公网安备35020302034844号
