Qt/PySide6：使用 QThread 实现无限数据获取循环的最佳方法是什么？

Question

我正在使用 PySide6 实现一个小型多线程 GUI 应用程序，以从（USB 连接的）传感器获取数据并使用 Qt 可视化数据。即，用户可以启动和停止数据获取：

单击播放按钮时，将创建一个工作对象并将其移动到

QThread

，然后启动

QThread

。互联网上充满了如何实现这种不确定（数据获取循环）的不同方法。这是迄今为止我最常遇到的两种主要方法：

• 方法 1（无限但用户可中断循环加上

  sleep()

  ）：

class Worker(QObject):
    def __init__(self, user_stop_event: Event)
        self.user_stop_event = user_stop_event
    
     def run(self):
         while not user_stop_event.isSet():
             self.fetch_data()
             # Process received signals:
             Qtcore.QApplication.processEvents()
             # A sleep is important since in Python threads cannot run really in parallel (on multicore systems) and without sleep we would block the main (GUI) thread!
             QThread.sleep(50)

     def fetch_data(self):
         ...

• 方法2（基于定时器的方法）：

class Worker(QObject):
    def __init__(self):
         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 的官方Threading Basics 网站在这里没有给出明确的建议。两者都在我这边工作，但我不太确定后续项目应使用哪一个作为我们的默认实现。

Answer 1

QT 在这方面确实没有发言权，这取决于你使用什么 API 进行通信的通信模式，以及是否支持同步或异步 IO。

如果通信是同步的，例如pyserial，那么你只需要在不睡眠的情况下执行

Approach1

，但在端口上添加超时，从同步IO读取通常会丢弃GIL，因此你不需要睡眠。 （它会使用 pyserial 删除 GIL，请检查任何其他 API 的文档）

class Worker(QObject):
    def __init__(self, user_stop_event: Event)
        self.user_stop_event = user_stop_event
    
     def run(self):
         while not user_stop_event.isSet():
             self.fetch_data()
             # Process received signals:
             Qtcore.QApplication.processEvents()

     def fetch_data(self):
         # call Serial.Read() here

但是，您可以从像 QTcpSocket 这样的异步 IO 中读取数据，然后您就可以

Approach2

因为 Qt 的事件循环在不执行 python 代码时会删除 GIL，所以在任何一种情况下都不需要休眠

class Worker(QObject):
    def __init__(self):
         self.socket = QTcpSocket(self)
         self.socket.readyRead.connect(self.fetch_data)
         self.socket.connectToHost(...)
         self.socket.waitForConnected()
         self.socket.write(...)

     def fetch_data(self):
         ...