PyQt5自定义QThread兼顾事件处理与串口操作的优化方案咨询

QThread的两种使用方式 #

• QThread实例运行默认事件循环，可通过信号与Worker类（以方法为槽）将事件分派至线程执行。
• 子类化QThread并重写run()执行自定义代码，但无法接收事件。

需求场景 #

现有SerialThread需同时实现两个核心功能：

• 定期检查串口并接收数据
• 支持主线程通过该线程发送数据（同一物理串口无法被两个serial对象同时使用）

现有基础实现 #

class SerialThread(QThread):
    
    class Worker(QObject):
        def __init__(self, n: int):
            super().__init__()
            self.n = n
        
        @pyqtSlot()
        def work(self):
            print(f"Exec from worker {self.n}...")
            print("Work Thread ID:", format(threading.get_ident(), '04X'))
    
    
    sig_new_data = pyqtSignal(list)
    _sig_work = pyqtSignal()
    
    _worker = Worker(1)
    
    _serial = serial.Serial()
    
    def __init__(self):
        super().__init__()
        self._worker.moveToThread(self)
        self._sig_work.connect(self._worker.work)
    
    def run(self):
        print("Serial Thread ID:", format(threading.get_ident(), '04X'))
        
        self.__open_com_port()
        
        while True:
            self.eventDispatcher().processEvents(QEventLoop.AllEvents)    # 处理待执行事件
            
            if not self._serial.is_open and not self.__open_com_port():
                self.msleep(2000)
                continue
            
            if self.__get_data():
                self.sig_new_data.emit()

    def queue_work(self):    # 主线程调用
        self._sig_work.emit()   # 向SerialThread事件循环添加待执行事件

当前存在的问题 #

1. 需要支持多种Worker类型或接收可调用对象的Worker类，尝试每次创建新Worker实例并临时连接信号，但函数内创建的实例会因垃圾回收失效，仅类成员或全局变量的Worker实例能正常工作：

def queue_work(thread: SerialThread):
    worker = SerialThread.Worker()
    worker.moveToThread(thread)
    thread._sig_work.connect(worker.work)
    thread._sig_work.emit() 
    thread._sig_work.disconnect(worker.work)

2. 未来多线程访问串口时，单一信号会引发冲突；且pyqtSignal必须定义在QObject子类中，无法在queue_work()内动态创建新信号。

寻求优化方案 #

希望得到更简洁优雅的实现方案，或解决多serial对象访问同一串口的替代方案。

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