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(func=func2, a=1, b=2) def func1(): time.sleep(1) print("func1 running") def func2(a, b): time.sleep(1) print(f"func2 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(f'func running: a={a}, b={b}') class MyThread(QThread): def __init__(self, func=None, *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, func=None, *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())加入了可变参数和停止线程的功能。