QtPy（PySide6），在线程中使用QEventLoop实现低成本待机

需求在主线程中启动一个功能函数在指定线程中运行功能函数运行完成后指定线程并不立即结束等待运行下一个功能函数。1、使用moveToThread创建QObject类将功能函数包装为QObject类的槽函数然后使用使用moveToThread移动到QObject类指定线程中运行使用信号触发功能函数已包装为QObject类的槽函数功能函数就会在指定线程内运行。import sys import time from PySide6.QtCore import QThread, Signal, QObject, Slot from PySide6.QtWidgets import QApplication, QPushButton # 功能函数 def func(): time.sleep(3) print(func running) # 定义一个QObject类 class FuncObj(QObject): run_signal Signal() def __init__(self): super().__init__() self.run_signal.connect(self.function) Slot() def function(self): func() if __name__ __main__: app QApplication(sys.argv) btn QPushButton(qwe) btn.show() t QThread() # 在主线程中创建线程 t func_obj FuncObj() # 创建函数对象 func_obj.moveToThread(t) # 将函数对象移动到线程 t 中函数对象的槽函数就会在线程 t 中运行 func_obj.run_signal.emit() # 发射run_signal信号触发函数对象的的槽函数 t.start() # 启动线程 app.exec()这是Qt的推荐用法。moveToThread方法需要为每一个功能函数创建QObject类对象并使用信号触发功能函数。如果QObject类对象是临时的运行完成后要对其进行销毁避免内存泄漏。进阶在任意线程普通调用QObject的函数不发射信号然后在QObject的内部发射信号使其运行在指定线程import sys import time from PySide6.QtCore import QThread, QEventLoop, QObject, Signal from PySide6.QtWidgets import QApplication, QWidget, QVBoxLayout, QPushButton class TaskObject(QObject): _signal_run Signal() def __init__(self): super().__init__() self.func None self.args None self.kwargs None self._signal_run.connect(self._on_signal_run) def _on_signal_run(self): if self.func: self.func(*self.args, **self.kwargs) else: print(no task func) def run_task(self, func, *args, **kwargs): self.func func self.args args self.kwargs kwargs self._signal_run.emit() def on_btn1(): task.run_task(func1) def on_btn2(): task.run_task(funcfunc2, a1, b2) def func1(): time.sleep(1) print(func1 running) def func2(a, b): time.sleep(1) print(ffunc2 running, a {a}, b {b}) if __name__ __main__: app QApplication(sys.argv) t QThread() # 在主线程中创建线程 t task TaskObject() task.moveToThread(t) t.start() widget QWidget() layout QVBoxLayout(widget) btn1 QPushButton(run func1) btn2 QPushButton(run func2) btn1.clicked.connect(on_btn1) btn2.clicked.connect(on_btn2) layout.addWidget(btn1) layout.addWidget(btn2) widget.show() app.exec()2、不使用moveToThread方法import sys import time from PySide6.QtCore import QThread from PySide6.QtWidgets import QApplication, QPushButton # 功能函数 def function(): time.sleep(3) print(func running) class MyThread(QThread): def __init__(self): super().__init__() self.func None self.start() def run(self): while 1: if self.func is None: continue self.func() self.func None def run_once(self, f): self.func f if __name__ __main__: app QApplication(sys.argv) thread MyThread() time.sleep(0.1) thread.run_once(function) btn QPushButton(验证是否在子线程中运行) btn.show() app.exec()这段代码在线程的run()函数中使用了循环等待功能函数的调用由于循环是持续运行的所以CPU的占用很高。3、使用QEventLoop实现低消耗待机import sys import threading import time from PySide6.QtCore import QThread, QEventLoop, Signal from PySide6.QtWidgets import QApplication, QPushButton, QWidget, QHBoxLayout # 功能函数 def function(): time.sleep(3) print(func running) class MyThread(QThread): def __init__(self): super().__init__() self.start() def run(self): self.loop QEventLoop() # 实例化线程以后创建事件循环使它运行在本线程内 self.loop.exec() # 将线程在此阻塞等待self.loop.quit()退出阻塞 while 1: self.func() self.loop.exec() def run_once(self, f): self.func f self.loop.quit() if __name__ __main__: app QApplication(sys.argv) thread MyThread() time.sleep(0.1) thread.run_once(function) btn QPushButton(验证是否在子线程中运行) btn.show() app.exec()线程的run()中使用了exec()将线程阻塞等待执行功能函数exec()对CPU的占用极低。并且在不创建QObject对象和不使用信号槽的前提下实现了功能函数的异步调用。4、完善以后的代码import sys import time from PySide6.QtCore import QThread, QEventLoop, Qt from PySide6.QtWidgets import QApplication, QPushButton, QWidget, QHBoxLayout # 功能函数 def func(a, b): time.sleep(3) # 模拟耗时操作 print(ffunc running: a{a}, b{b}) class MyThread(QThread): def __init__(self, funcNone, *args, **kwargs): 可复用的QThread线程类支持触发执行指定函数 :param func: 线程内运行的函数 :param args: 函数位置参数 :param kwargs: 函数关键字参数 super().__init__() self._func func # 待执行函数 self._args args # 位置参数 self._kwargs kwargs # 关键字参数 self._running True # 线程运行标识 self._loop None # 事件循环对象延迟初始化 def run(self): 线程核心运行逻辑 # 在run方法内初始化事件循环确保属于当前线程 self._loop QEventLoop() # 循环等待触发直到停止线程 while self._running: # 阻塞等待触发quit()会解除阻塞 self._loop.exec() # 解除阻塞后若线程仍在运行且有函数则执行 if self._running and self._func: try: self._func(*self._args, **self._kwargs) except Exception as e: print(f函数执行出错: {e}) def run_once(self, funcNone, *args, **kwargs): 触发线程执行一次函数 :param func: 可选替换要执行的函数 :param args: 可选替换位置参数 :param kwargs: 可选替换关键字参数 # 如果传入了新的函数/参数更新 if func: self._func func self._args args self._kwargs kwargs # 确保线程和事件循环有效 if self.isRunning() and self._loop and self._running: self._loop.quit() # 解除事件循环阻塞触发函数执行 else: print(线程未运行或已停止无法执行) def stop(self): 安全停止线程并释放资源 self._running False # 标记线程停止 if self._loop: self._loop.quit() # 解除事件循环阻塞 self.wait(3000) # 等待线程结束最多3秒 # 线程停止后清理资源 self._loop None self._func None if __name__ __main__: app QApplication(sys.argv) # 创建线程初始绑定func和参数1,2不立即执行 thread MyThread(func, 1, 2) thread.start() # 手动启动线程推荐显式调用 # 创建UI界面 widget QWidget() widget.setWindowTitle(QThread复用示例) layout QHBoxLayout(widget) # 按钮1使用初始参数执行一次 btn1 QPushButton(初始参数执行) btn1.clicked.connect(thread.run_once) # 按钮2使用新参数执行一次 btn2 QPushButton(新参数执行) btn2.clicked.connect(lambda: thread.run_once(func, 3, 4)) # 按钮3停止线程 btn3 QPushButton(结束线程) btn3.clicked.connect(thread.stop) layout.addWidget(btn1) layout.addWidget(btn2) layout.addWidget(btn3) widget.resize(400, 100) widget.show() # 程序退出时确保线程停止 app.aboutToQuit.connect(thread.stop) sys.exit(app.exec())加入了可变参数和停止线程的功能。