PySide2中worker调用moveToThread后自定义信号关联槽函数无法接收信号问题

问题根因分析 #

1. 跨线程信号槽依赖事件循环运转
   默认跨线程的信号槽连接类型是Qt.QueuedConnection，这类信号会被投递到接收方所属线程的事件队列，等待线程的事件循环处理。原代码中发完信号就结束函数，主线程没有保持运行，子线程的事件循环还没来得及处理队列里的信号，函数就退出了，自然收不到信号。
2. 局部变量提前被垃圾回收
   原代码中定义的communicate、worker、my_thread都是moveToThreadTest函数的局部变量，函数执行结束后这些对象没有被其他引用持有，会被Python垃圾回收，信号和连接直接失效，这也是debug模式下连process槽都触发不了的核心原因：debug模式下GC触发时机更早，子线程还没来得及处理started信号，对象就被销毁了。
3. 信号发送时机早于子线程事件循环启动
   原代码调用my_thread.start()之后立刻发射自定义信号，此时子线程的事件循环可能还没完成初始化，信号投递到空队列直接丢失。

修复方案 #

需要做三处修改：

• 初始化QApplication保持主事件循环运行，避免函数退出后局部变量被回收
• 调整信号发送时机，在子线程事件循环启动完成后再发射测试信号
• 所有相关对象（worker、communicate、QThread实例）要持有强引用，避免被GC回收

修复后可运行代码 #

import sys
import threading
from PySide2.QtCore import QObject, Signal, Slot, QThread, Qt
from loguru import logger

class DbWorker(QObject):
    def __init__(self):
        super().__init__()
        logger.info(f"{threading.get_ident()}: DbWorker instantiated...")

    @Slot(str)
    def test_slot_str(self, test_text: str):
        logger.info(f"{threading.get_ident()}: test_slot_str fired. Got Message '{test_text}'")

    @Slot()
    def test_slot_empty(self):
        logger.info(f"{threading.get_ident()}: test_slot_empty fired ")

    @Slot()
    def process(self):
        logger.info(f"{threading.get_ident()}: Process slot activated!")

class Communicate(QObject):
    signal_str = Signal(str)
    signal_empty = Signal()

if __name__ == "__main__":
    from PySide2.QtWidgets import QApplication
    # 必须初始化QApplication，提供全局事件循环支持
    app = QApplication(sys.argv)

    logger.info(f"{threading.get_ident()}: main started...")
    # 将对象提升为全局作用域持有，避免被GC回收
    worker = DbWorker()
    communicate = Communicate()
    my_thread = QThread()
    worker.moveToThread(my_thread)

    communicate.signal_str.connect(worker.test_slot_str, Qt.QueuedConnection)
    communicate.signal_empty.connect(worker.test_slot_empty, Qt.QueuedConnection)
    my_thread.started.connect(worker.process)
    # 子线程启动完成后再发送测试信号，确保事件循环正常运转
    def send_test_signals():
        communicate.signal_str.emit("test")
        communicate.signal_empty.emit()
        # 测试完成后安全退出线程和主事件循环
        my_thread.quit()
        my_thread.wait()
        app.quit()
    my_thread.started.connect(send_test_signals)
    
    my_thread.start()

    # 启动主事件循环，保持程序运行
    sys.exit(app.exec_())

运行输出示例 #

- INFO     [test.py:12 - __init__] 140735223237440: main started...
 - INFO     [test.py:18 - __init__] 140735223237440: DbWorker instantiated...
 - INFO     [test.py:30 - process] 123145529815040: Process slot activated!
 - INFO     [test.py:22 - test_slot_str] 123145529815040: test_slot_str fired. Got Message 'test'
 - INFO     [test.py:26 - test_slot_empty] 123145529815040: test_slot_empty fired 

额外注意事项 #

• 不要手动删除已经move到子线程的worker对象，如果要销毁，先调用my_thread.quit()等待线程退出后再处理
• 跨线程连接不要指定Qt.DirectConnection，否则槽函数会在发送信号的线程执行，无法实现子线程执行逻辑的需求
• PySide/PyQt的所有QObject派生类实例如果要用到信号槽，必须保证实例有至少一个强引用持有，否则会被Python GC回收，连接直接失效

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