PyQt5中MainWindow激活时如何周期性调用函数？Multiprocessing报错求解

错误原因 #

multiprocessing启动子进程时，需要将目标函数和相关参数序列化（pickle）后传递给子进程。但Qt的MainWindow对象包含底层C++绑定的UI资源，这类对象无法被pickle序列化。另外，Qt的UI操作必须在主线程执行，子进程直接操作主进程的UI对象本身就是错误的做法。

方案1：用Qt原生QTimer实现周期性任务（推荐） #

如果你的周期性任务不需要耗时操作，直接用Qt的QTimer是最简单的方案，完全符合Qt的线程模型：

import time
from PyQt5 import QtWidgets, QtCore


class MainWindow(QtWidgets.QWidget):
    def __init__(self):
        super().__init__()
        self.initializeUI()
        self.show()

    def initializeUI(self):
        self.setLayout(QtWidgets.QGridLayout())
        dummy_btn = QtWidgets.QPushButton("Ok")
        self.layout().addWidget(dummy_btn)

        # 创建QTimer实现周期性调用
        self.updater_timer = QtCore.QTimer(self)
        self.updater_timer.timeout.connect(self.update_function)
        self.updater_timer.start(2000)  # 每隔2秒触发一次

    def update_function(self):
        print("This text, again!")


app = QtWidgets.QApplication([])
mw = MainWindow()
app.exec_()

方案2：多进程处理耗时任务（如果任务耗时） #

如果update_function是耗时操作，不能阻塞主线程，就需要用多进程处理任务，再通过进程间通信把结果传给主线程，由主线程更新UI：

import time
import multiprocessing
from PyQt5 import QtWidgets, QtCore


def worker(queue):
    """子进程的任务函数"""
    while True:
        time.sleep(2)
        queue.put("This text, again!")  # 把结果放入队列


class MainWindow(QtWidgets.QWidget):
    def __init__(self):
        super().__init__()
        self.queue = multiprocessing.Queue()
        self.initializeUI()
        self.start_worker()
        self.show()

    def initializeUI(self):
        self.setLayout(QtWidgets.QGridLayout())
        dummy_btn = QtWidgets.QPushButton("Ok")
        self.layout().addWidget(dummy_btn)

        # 用QTimer定期检查队列是否有新数据
        self.check_queue_timer = QtCore.QTimer(self)
        self.check_queue_timer.timeout.connect(self.process_queue)
        self.check_queue_timer.start(100)  # 每100毫秒检查一次

    def start_worker(self):
        self.worker_process = multiprocessing.Process(target=worker, args=(self.queue,))
        self.worker_process.daemon = True  # 主进程退出时子进程自动结束
        self.worker_process.start()

    def process_queue(self):
        """处理子进程传来的数据"""
        while not self.queue.empty():
            message = self.queue.get()
            print(message)
            # 如果需要更新UI，在这里操作（比如self.label.setText(message)）


app = QtWidgets.QApplication([])
mw = MainWindow()
app.exec_()

关键注意点 #

• 永远不要在子进程中操作Qt的UI元素，所有UI更新必须在主线程完成。
• QTimer是Qt推荐的周期性任务实现方式，避免了多进程/多线程的序列化问题。
• 多进程通信优先用Queue或Pipe，不要直接传递Qt对象。

内容的提问来源于stack exchange，提问作者user11794063