当前位置： 文江博客话题详情

PyQT5：如何使用相同的工作定义在单独的 QThread 中运行任意 Callable？

发布于 2025-01-10 08:29:25 字数 5249 浏览 0 评论 0原文

如何定义一个工作线程以便我可以在单独的 QThread 中运行任意 Callable ？

我尝试使用 lambda 表达式在运行时传递参数，但这不起作用（工作线程仍然在主线程中运行），请参阅下面的代码，方法 run_in_separate_thread

import sys
from time import sleep
from typing import Callable

from PyQt5.QtWidgets import (
    QApplication,
    QMainWindow,
    QPushButton,
    QVBoxLayout,
    QWidget, QPlainTextEdit,
)

from PyQt5.QtCore import QObject, QThread, pyqtSignal


class Window(QMainWindow):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.setupUi()

    def setupUi(self):
        self.resize(300, 150)
        self.centralWidget = QWidget()
        self.setCentralWidget(self.centralWidget)
        self.text_edit = QPlainTextEdit()
        self.bt = QPushButton("Run in separate thread", self)
        layout = QVBoxLayout()
        layout.addWidget(self.text_edit)
        layout.addWidget(self.bt)
        self.centralWidget.setLayout(layout)

        self.bt.clicked.connect(lambda: self.run_in_separate_thread(my_long_function))

    def report_progress(self, text: str) -> None:
        """ Report progress of function in separate thread """
        self.text_edit.appendPlainText(str(text))

    def run_in_separate_thread(self, function: Callable) -> None:
        """ Wrapper to run a function in a separate thread """

        self.thread = QThread(objectName="workerThread")
        self.worker = Worker()
        self.worker.moveToThread(self.thread)
        self.worker.finished.connect(self.thread.quit)
        self.worker.finished.connect(self.worker.deleteLater)
        self.thread.finished.connect(self.thread.deleteLater)
        self.worker.progress.connect(self.report_progress)

        self.thread.started.connect(lambda: self.worker.run(function))
        self.thread.start()


class Worker(QObject):
    finished = pyqtSignal()
    progress = pyqtSignal(str)

    def run(self, function: Callable):
        function(self.progress)
        self.finished.emit()


def my_long_function(signal):
    print(f"mylongfunction thread: {QThread.currentThread().objectName(), int(QThread.currentThreadId())}")
    for _ in range(5):
        sleep(1)
        signal.emit("hello")


app = QApplication(sys.argv)
QThread.currentThread().setObjectName('main')
print(f"Main thread: {QThread.currentThread().objectName(), int(QThread.currentThreadId())}")
win = Window()
win.show()
sys.exit(app.exec())

：给我以下输出，并且 QTextEdit 中没有显示“实时”输出。

Main thread: ('main', 139754959259456)
mylongfunction thread: ('main', 139754959259456)

是因为 lambda 表达式在主线程中“存活”吗？ 如果我在工作 run() 方法中硬编码 my_long_function 并删除所有 lambda 表达式，它会按预期工作，请参见下面的代码：

import sys
from time import sleep
from typing import Callable

from PyQt5.QtWidgets import (
    QApplication,
    QMainWindow,
    QPushButton,
    QVBoxLayout,
    QWidget, QPlainTextEdit,
)

from PyQt5.QtCore import QObject, QThread, pyqtSignal


class Window(QMainWindow):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.setupUi()

    def setupUi(self):
        self.resize(300, 150)
        self.centralWidget = QWidget()
        self.setCentralWidget(self.centralWidget)
        self.text_edit = QPlainTextEdit()
        self.bt = QPushButton("Run in separate thread", self)
        layout = QVBoxLayout()
        layout.addWidget(self.text_edit)
        layout.addWidget(self.bt)
        self.centralWidget.setLayout(layout)

        self.bt.clicked.connect(self.run_in_separate_thread)

    def report_progress(self, text: str) -> None:
        """ Report progress of function in separate thread """
        self.text_edit.appendPlainText(str(text))

    def run_in_separate_thread(self) -> None:
        """ Wrapper to run a function in a separate thread """

        self.thread = QThread(objectName="workerThread")
        self.worker = Worker()
        self.worker.moveToThread(self.thread)
        self.worker.finished.connect(self.thread.quit)
        self.worker.finished.connect(self.worker.deleteLater)
        self.thread.finished.connect(self.thread.deleteLater)
        self.worker.progress.connect(self.report_progress)

        self.thread.started.connect(self.worker.run)
        self.thread.start()


class Worker(QObject):
    finished = pyqtSignal()
    progress = pyqtSignal(str)

    def run(self):
        self.my_long_function()
        self.finished.emit()

    def my_long_function(self):
        print(f"mylongfunction thread: {QThread.currentThread().objectName(), int(QThread.currentThreadId())}")
        for _ in range(5):
            sleep(1)
            self.progress.emit("hello")


def my_long_function(signal):
    print(f"mylongfunction thread: {QThread.currentThread().objectName(), int(QThread.currentThreadId())}")
    for _ in range(5):
        sleep(1)
        signal.emit("hello")


app = QApplication(sys.argv)
QThread.currentThread().setObjectName('main')
print(f"Main thread: {QThread.currentThread().objectName(), int(QThread.currentThreadId())}")
win = Window()
win.show()
sys.exit(app.exec())

这会给出以下所需的输出：

Main thread: ('main', 139687209740096)
mylongfunction thread: ('workerThread', 139686815192832)
原文

How can I define a worker so that I can run an arbitrary Callable in a separate QThread?

I tried using lambda expression to pass arguments at run-time, but this does not work (worker still runs in the main thread), see code below, method run_in_separate_thread:

import sys
from time import sleep
from typing import Callable

from PyQt5.QtWidgets import (
    QApplication,
    QMainWindow,
    QPushButton,
    QVBoxLayout,
    QWidget, QPlainTextEdit,
)

from PyQt5.QtCore import QObject, QThread, pyqtSignal


class Window(QMainWindow):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.setupUi()

    def setupUi(self):
        self.resize(300, 150)
        self.centralWidget = QWidget()
        self.setCentralWidget(self.centralWidget)
        self.text_edit = QPlainTextEdit()
        self.bt = QPushButton("Run in separate thread", self)
        layout = QVBoxLayout()
        layout.addWidget(self.text_edit)
        layout.addWidget(self.bt)
        self.centralWidget.setLayout(layout)

        self.bt.clicked.connect(lambda: self.run_in_separate_thread(my_long_function))

    def report_progress(self, text: str) -> None:
        """ Report progress of function in separate thread """
        self.text_edit.appendPlainText(str(text))

    def run_in_separate_thread(self, function: Callable) -> None:
        """ Wrapper to run a function in a separate thread """

        self.thread = QThread(objectName="workerThread")
        self.worker = Worker()
        self.worker.moveToThread(self.thread)
        self.worker.finished.connect(self.thread.quit)
        self.worker.finished.connect(self.worker.deleteLater)
        self.thread.finished.connect(self.thread.deleteLater)
        self.worker.progress.connect(self.report_progress)

        self.thread.started.connect(lambda: self.worker.run(function))
        self.thread.start()


class Worker(QObject):
    finished = pyqtSignal()
    progress = pyqtSignal(str)

    def run(self, function: Callable):
        function(self.progress)
        self.finished.emit()


def my_long_function(signal):
    print(f"mylongfunction thread: {QThread.currentThread().objectName(), int(QThread.currentThreadId())}")
    for _ in range(5):
        sleep(1)
        signal.emit("hello")


app = QApplication(sys.argv)
QThread.currentThread().setObjectName('main')
print(f"Main thread: {QThread.currentThread().objectName(), int(QThread.currentThreadId())}")
win = Window()
win.show()
sys.exit(app.exec())

This gives me the following output, and no "live" output is shown in the QTextEdit.

Main thread: ('main', 139754959259456)
mylongfunction thread: ('main', 139754959259456)

Is it because lambda expression "live" in the Main thread?
If I hard code my_long_function in the worker run() method and remove all lambdas expressions, it works as intended, see code below:

import sys
from time import sleep
from typing import Callable

from PyQt5.QtWidgets import (
    QApplication,
    QMainWindow,
    QPushButton,
    QVBoxLayout,
    QWidget, QPlainTextEdit,
)

from PyQt5.QtCore import QObject, QThread, pyqtSignal


class Window(QMainWindow):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.setupUi()

    def setupUi(self):
        self.resize(300, 150)
        self.centralWidget = QWidget()
        self.setCentralWidget(self.centralWidget)
        self.text_edit = QPlainTextEdit()
        self.bt = QPushButton("Run in separate thread", self)
        layout = QVBoxLayout()
        layout.addWidget(self.text_edit)
        layout.addWidget(self.bt)
        self.centralWidget.setLayout(layout)

        self.bt.clicked.connect(self.run_in_separate_thread)

    def report_progress(self, text: str) -> None:
        """ Report progress of function in separate thread """
        self.text_edit.appendPlainText(str(text))

    def run_in_separate_thread(self) -> None:
        """ Wrapper to run a function in a separate thread """

        self.thread = QThread(objectName="workerThread")
        self.worker = Worker()
        self.worker.moveToThread(self.thread)
        self.worker.finished.connect(self.thread.quit)
        self.worker.finished.connect(self.worker.deleteLater)
        self.thread.finished.connect(self.thread.deleteLater)
        self.worker.progress.connect(self.report_progress)

        self.thread.started.connect(self.worker.run)
        self.thread.start()


class Worker(QObject):
    finished = pyqtSignal()
    progress = pyqtSignal(str)

    def run(self):
        self.my_long_function()
        self.finished.emit()

    def my_long_function(self):
        print(f"mylongfunction thread: {QThread.currentThread().objectName(), int(QThread.currentThreadId())}")
        for _ in range(5):
            sleep(1)
            self.progress.emit("hello")


def my_long_function(signal):
    print(f"mylongfunction thread: {QThread.currentThread().objectName(), int(QThread.currentThreadId())}")
    for _ in range(5):
        sleep(1)
        signal.emit("hello")


app = QApplication(sys.argv)
QThread.currentThread().setObjectName('main')
print(f"Main thread: {QThread.currentThread().objectName(), int(QThread.currentThreadId())}")
win = Window()
win.show()
sys.exit(app.exec())

This gives following desired output:

Main thread: ('main', 139687209740096)
mylongfunction thread: ('workerThread', 139686815192832)

如果你对这篇内容有疑问，欢迎到本站社区发帖提问 参与讨论，获取更多帮助，或者扫码二维码加入 Web 技术交流群。

扫码二维码加入Web技术交流群

发布评论

需要 登录 才能够评论， 你可以免费 注册 一个本站的账号。

评论（1）

蓝海 2025-01-17 08:29:25

好的，这里有一个解决方案，通过将 Callable 定义为工作线程的实例变量：

这样，工作线程就可以在单独的线程中正确运行。

完整代码：

import sys
from time import sleep
from typing import Callable

from PyQt5.QtWidgets import (
    QApplication,
    QMainWindow,
    QPushButton,
    QVBoxLayout,
    QWidget, QPlainTextEdit,
)

from PyQt5.QtCore import QObject, QThread, pyqtSignal


class Window(QMainWindow):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.setupUi()

    def setupUi(self):
        self.resize(300, 150)
        self.centralWidget = QWidget()
        self.setCentralWidget(self.centralWidget)
        self.text_edit = QPlainTextEdit()
        self.bt = QPushButton("Run in separate thread", self)
        layout = QVBoxLayout()
        layout.addWidget(self.text_edit)
        layout.addWidget(self.bt)
        self.centralWidget.setLayout(layout)

        self.bt.clicked.connect(lambda: self.run_in_separate_thread(my_long_function))


    def report_progress(self, text: str) -> None:
        """ Report progress of function in separate thread """
        self.text_edit.appendPlainText(str(text))

    def run_in_separate_thread(self, f: Callable) -> None:
        """ Wrapper to run a function in a separate thread """

        self.thread = QThread(objectName="workerThread")
        self.worker = Worker()

        self.worker.moveToThread(self.thread)
        self.worker.finished.connect(self.thread.quit)
        self.worker.finished.connect(self.worker.deleteLater)
        self.thread.finished.connect(self.thread.deleteLater)
        self.worker.progress.connect(self.report_progress)
        self.thread.started.connect(self.worker.run)

        self.worker.task = f
        self.thread.start()


class Worker(QObject):
    finished = pyqtSignal()
    progress = pyqtSignal(str)
    task = None

    def run(self):
        self.task(self.progress)
        self.finished.emit()


def my_long_function(signal):
    print(f"mylongfunction thread: {QThread.currentThread().objectName(), int(QThread.currentThreadId())}")
    for _ in range(5):
        sleep(1)
        signal.emit("hello")


app = QApplication(sys.argv)
QThread.currentThread().setObjectName('main')
print(f"Main thread: {QThread.currentThread().objectName(), int(QThread.currentThreadId())}")
win = Window()
win.show()
sys.exit(app.exec())

OK here is a solution, by defining the Callable as an instance variable of the worker:

With this the worker is properly running in a separate thread.

full code:

import sys
from time import sleep
from typing import Callable

from PyQt5.QtWidgets import (
    QApplication,
    QMainWindow,
    QPushButton,
    QVBoxLayout,
    QWidget, QPlainTextEdit,
)

from PyQt5.QtCore import QObject, QThread, pyqtSignal


class Window(QMainWindow):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.setupUi()

    def setupUi(self):
        self.resize(300, 150)
        self.centralWidget = QWidget()
        self.setCentralWidget(self.centralWidget)
        self.text_edit = QPlainTextEdit()
        self.bt = QPushButton("Run in separate thread", self)
        layout = QVBoxLayout()
        layout.addWidget(self.text_edit)
        layout.addWidget(self.bt)
        self.centralWidget.setLayout(layout)

        self.bt.clicked.connect(lambda: self.run_in_separate_thread(my_long_function))


    def report_progress(self, text: str) -> None:
        """ Report progress of function in separate thread """
        self.text_edit.appendPlainText(str(text))

    def run_in_separate_thread(self, f: Callable) -> None:
        """ Wrapper to run a function in a separate thread """

        self.thread = QThread(objectName="workerThread")
        self.worker = Worker()

        self.worker.moveToThread(self.thread)
        self.worker.finished.connect(self.thread.quit)
        self.worker.finished.connect(self.worker.deleteLater)
        self.thread.finished.connect(self.thread.deleteLater)
        self.worker.progress.connect(self.report_progress)
        self.thread.started.connect(self.worker.run)

        self.worker.task = f
        self.thread.start()


class Worker(QObject):
    finished = pyqtSignal()
    progress = pyqtSignal(str)
    task = None

    def run(self):
        self.task(self.progress)
        self.finished.emit()


def my_long_function(signal):
    print(f"mylongfunction thread: {QThread.currentThread().objectName(), int(QThread.currentThreadId())}")
    for _ in range(5):
        sleep(1)
        signal.emit("hello")


app = QApplication(sys.argv)
QThread.currentThread().setObjectName('main')
print(f"Main thread: {QThread.currentThread().objectName(), int(QThread.currentThreadId())}")
win = Window()
win.show()
sys.exit(app.exec())
回复 原文
~没有更多了~

关于作者

对不⑦

暂无简介

文章
评论
26 人气
发私信

相关话题

更多

热门标签

操作系统 程序设计 IT运维 Linux系统管理 JavaScript 服务器应用 solaris C/C++ PHP Shell BSD Vue.js aix Oracle Python HTML 系统管理 HTML5 CSS 前端
更多

推荐作者

紫罗兰の梦幻

文章 0 评论 0

-2134

文章 0 评论 0

liuxuanli

文章 0 评论 0

意中人

文章 0 评论 0

○愚か者の日

文章 0 评论 0

xxhui

文章 0 评论 0

更多

友情链接

文江博客
    我们使用 Cookies 和其他技术来定制您的体验包括您的登录状态等。通过阅读我们的 隐私政策 了解更多相关信息。 单击 接受 或继续使用网站，即表示您同意使用 Cookies 和您的相关数据。
    原文