PyQt6中QThread的started信号连接带参函数失败的解决方法

当你把QThread的started信号连接到lambda调用带参数的worker.run()时，代码会在主线程执行（线程ID和应用主线程一致），但连接无参函数却能正常在子线程运行。这是因为lambda的调用上下文是主线程，直接调用worker.run()属于普通函数调用，并没有触发PyQt的子线程调度机制；而直接绑定worker的无参方法时，PyQt会自动将方法执行调度到worker所在的子线程。

下面是三种可行的解决方案：

方法1：用functools.partial包装带参数方法 #

functools.partial可以帮你把参数绑定到目标方法上，同时让PyQt正确识别这是要在worker所在线程执行的方法。

修改test方法中的连接代码：

from functools import partial

# 替换原lambda连接
self.threader.started.connect(partial(self.worker.run, stop=10))  # 按需指定参数值

方法2：给Worker类添加带参数的自定义信号 #

通过自定义信号传递参数，让信号触发run方法执行，确保代码在子线程运行：

首先修改Worker类，新增带参数的信号并绑定到run方法：

class Worker(QObject):
    send = pyqtSignal(str)
    finished = pyqtSignal()
    start_run = pyqtSignal(int)  # 新增带int参数的信号

    def __init__(self) -> None:
        super().__init__()
        self.start_run.connect(self.run)  # 信号与run方法绑定

    # 原run方法保持不变
    def run(self, stop = 5):
        print(f"Worker thread: {int(self.thread().currentThreadId())}")
        for i in range(stop):
            time.sleep(1)
            self.send.emit(str(i))
        self.finished.emit()

然后在test方法中，将started信号连接到自定义信号的发射：

# 替换原lambda连接
self.threader.started.connect(lambda: self.worker.start_run.emit(10))  # 传递参数10

方法3：用QMetaObject.invokeMethod异步调用 #

通过QMetaObject.invokeMethod显式指定在worker所在线程调用run方法，确保执行上下文正确：

修改test方法中的连接代码：

from PyQt6.QtCore import QMetaObject, Qt

# 替换原lambda连接
self.threader.started.connect(
    lambda: QMetaObject.invokeMethod(
        self.worker,
        "run",
        Qt.ConnectionType.QueuedConnection,  # 确保在子线程事件循环中执行
        Qt.Q_ARG(int, 10)  # 传递参数，需匹配方法参数类型
    )
)

完整示例代码（方法1） #

from PyQt6.QtWidgets import QLabel, QApplication, QHBoxLayout, QPushButton, QWidget
from PyQt6.QtCore import pyqtSignal, QObject, QThread
from functools import partial
import time
import sys

class Worker(QObject):
    send = pyqtSignal(str)
    finished = pyqtSignal()
    def __init__(self) -> None:
        super().__init__()

    def run(self, stop = 5):
        print(f"Worker thread: {int(self.thread().currentThreadId())}")
        for i in range(stop):
            time.sleep(1)
            self.send.emit(str(i))
        self.finished.emit()
    
    def runNoParam(self):
        print(f"Worker thread: {int(self.thread().currentThreadId())}")
        for i in range(5):
            time.sleep(1)
            self.send.emit(str(i))
        self.finished.emit()

class Window(QWidget):
    def __init__(self) -> None:
        super().__init__()
        layout = QHBoxLayout()
        self.label = QLabel("number")
        self.btn = QPushButton("click me")
        layout.addWidget(self.label)
        layout.addWidget(self.btn)
        self.setLayout(layout)
        self.btn.clicked.connect(self.test)


    def test(self):
        self.threader = QThread()
        self.worker = Worker()
        self.worker.moveToThread(self.threader)

        # 使用functools.partial连接带参数的run方法
        self.threader.started.connect(partial(self.worker.run, stop=10))
        
        self.worker.send.connect(lambda s: self.label.setText(s))
        
        self.worker.finished.connect(self.threader.quit)
        self.worker.finished.connect(self.threader.deleteLater)
        self.threader.finished.connect(self.worker.deleteLater)  # 可选：清理worker

        self.threader.start()

if __name__ == '__main__':
    app = QApplication(sys.argv)
    win = Window()
    win.show()
    print(f"Application thread: {int(QApplication.instance().thread().currentThreadId())}")
    sys.exit(app.exec())

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