PySide6中基于QThread实现无限数据获取循环的最优方案探讨

我正在使用PySide6开发一款小型多线程GUI应用，用于从USB连接的传感器获取数据并通过Qt进行可视化，用户可启动或停止数据获取：

点击播放按钮时，会创建Worker对象并将其移至QThread，随后启动QThread。目前我了解到两种实现无限数据获取循环的主流方案：

方案1（可中断无限循环+sleep()） #

class Worker(QObject):
    def __init__(self, user_stop_event: Event):
        super().__init__()
        self.user_stop_event = user_stop_event
    
    def run(self):
        while not self.user_stop_event.isSet():
            self.fetch_data()
            # 处理接收的信号：
            QtCore.QApplication.processEvents()
            # 睡眠很重要，因为Python线程无法真正多核并行，无睡眠会阻塞主线程！
            QThread.sleep(50)

    def fetch_data(self):
        ...

方案2（基于定时器的实现） #

class Worker(QObject):
    def __init__(self):
        super().__init__()

        timer = QTimer()
        timer.timeout.connect(self.fetch_data)
        timer.start(50)

    def fetch_data(self):
        ...

两种方案使用相同的线程启动机制：

thread = QThread()
worker = Worker()
worker.moveToThread(thread)
thread.started.connect(worker.run)
...

问题 #

请问两种方案的优缺点是什么？哪种是更推荐的实现方式？

遗憾的是Qt官方的线程基础文档并未给出明确建议，两种方案均能正常运行，但我不确定后续项目应选用哪种作为默认实现。

方案1优缺点 #

优点：

• 逻辑直观，符合常规循环任务的思维模式，理解和调试成本低
• 对循环的控制更灵活，可在每次迭代中根据场景调整执行逻辑（比如动态修改采样间隔）
• 停止机制直接，通过user_stop_event能快速中断循环

缺点：

• 需手动调用QApplication.processEvents()才能处理线程内的信号槽，容易遗漏导致信号延迟或不响应
• QThread.sleep(50)是阻塞式睡眠，期间线程无法处理任何事件，灵活性不足
• 若fetch_data()执行时间不稳定，会导致循环间隔偏离预期，难以保证固定采样频率
• 依赖全局QApplication，代码耦合性较高

方案2优缺点 #

优点：

• 完全遵循Qt事件循环机制，无需手动处理事件，线程内信号槽可自动响应
• 定时器基于事件循环触发，即使fetch_data()有小幅延迟，后续触发时间会自动调整，能更稳定地保证固定采样间隔
• 代码结构简洁，契合Qt设计范式，无需维护额外的停止事件变量
• 停止操作更优雅，只需调用timer.stop()即可终止任务

缺点：

• 对于需根据前一次执行结果调整后续逻辑的复杂场景，定时器回调模式不如显式循环直观
• 原代码中timer为局部变量会被垃圾回收，需改为实例变量才能正常工作；且需确保Worker已移至目标线程后再启动定时器，否则定时器可能运行在主线程

推荐方案 #

优先选择方案2，理由如下：

1. 贴合Qt事件驱动模型，避免手动处理事件带来的潜在问题
2. 能更可靠地维持固定采样间隔，适合传感器数据采集这类对定时精度有要求的场景
3. 代码符合Qt最佳实践，后续维护和扩展更便捷

优化后的方案2代码示例：

class Worker(QObject):
    def __init__(self):
        super().__init__()
        self.timer = QTimer()  # 改为实例变量避免被回收
        self.timer.timeout.connect(self.fetch_data)
    
    def start(self):
        self.timer.start(50)
    
    def stop(self):
        self.timer.stop()
    
    def fetch_data(self):
        ...

同时调整线程启动逻辑，将thread.started连接到worker.start而非worker.run，更符合职责分离原则。

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