前言

关于PySide6/PyQT多线程系列的最后一篇。写这篇文章的动机是方便后续代码的直接复用。

本篇文章实际是水文，给出了 PySide6/PyQT的多线程以及线程池的基础使用模板，方便后面有需要时候直接拿来就用。

多线程

这里分两种情况来谈论，有返回值与无返回值。

无返回值

代码释义

定义 MyThread 的线程类，继承自 QThread；
在构造函数中，接收一个函数 func 和其它参数 *args 和 **kwargs；
并在 run() 方法中，调用传入的函数，并将参数传递给它。

定义 do_something() 的函数，它表示一个耗时的操作。

在主程序中，实例化MyThread 对象 thread，将 do_something 函数作为参数传递给它。
然后，通过调用 start() 方法启动线程，线程会自动执行 run() 方法中的任务。

# -*- coding: utf-8 -*-

import time
from PySide6.QtCore import (Qt, QThread)


# 自定义的工作任务类
class MyThread(QThread):
    def __init__(self, func, *args, **kwargs):
        super().__init__()
        self.func = func
        self.args = args
        self.kwargs = kwargs

    def run(self):
        self.func(*self.args, **self.kwargs)


def do_something():
    time.sleep(100)
    return '耗时操作'


if __name__ == '__main__':
    thread = MyThread(do_something)
    thread.start()

有返回值

有返回值，也分两种情况来讨论

（不推荐）使用线程对象的 wait() 方法等待子线程执行完毕，然后通过自定义的方法或属性来获取返回值
（推荐）使用信号和槽机制进行线程间通信

不推荐

这种情况是不推荐的，不够线程安全，
在多线程编程中，共享数据可能存在竞争条件和线程安全性问题，从而引发线程安全性问题。

# -*- coding: utf-8 -*-

import time
from PySide6.QtCore import (Qt, QThread)


# 自定义的工作任务类
class MyThread(QThread):
    def __init__(self, func, *args, **kwargs):
        super().__init__()
        self.res = str()
        self.func = func
        self.args = args
        self.kwargs = kwargs

    def run(self):
        self.res = self.func(*self.args, **self.kwargs)


def do_something():
    time.sleep(100)
    return '耗时操作'


if __name__ == '__main__':
    thread = MyThread(do_something)
    thread.start()
    thread.wait()
    print(thread.res)

推荐

代码释义

整体流程如下：

创建继承QThread 的 MyThead类，接收一个函数和不定长传参；重写了 run方法并在这里执行传递过来的函数和传参；
创建继承自 QWidget 的 MainWindow 类，用于显示窗口和处理事件；
在 setup_ui 方法中设置窗口的标题、大小，并创建标签和按钮，并将它们添加到垂直布局中；
在 setup_thread 方法中创建 MyThread 线程对象，并连接它的 finish 信号到槽函数 thread_finished；
在 do_something函数中模拟一个耗时的任务，并返回任务的结果；
在 thread_finished 槽函数中，接收到任务完成的信号后，将结果更新到标签上；
运行程序后，窗口会显示一个标签和一个按钮。点击按钮后，耗时任务会在后台执行，执行完成后，结果会显示在标签上。

# -*- coding: utf-8 -*-

import time

from PySide6.QtCore import (QThread, Signal, Slot, QSize)
from PySide6.QtWidgets import (QApplication, QPushButton, QLabel, QVBoxLayout, QWidget)


class MyThread(QThread):
    finish = Signal(str)

    def __init__(self, func, *args, **kwargs):
        super().__init__()
        self.func = func
        self.args = args
        self.kwargs = kwargs

    def run(self):
        res = self.func(*self.args, **self.kwargs)
        # 任务完成后发出信号
        self.finish.emit(res)


def do_something():
    time.sleep(1)
    return 'the function execution completed!'


class MainWindow(QWidget):
    def __init__(self, parent=None):
        super().__init__(parent=parent)
        self.setup_ui()
        self.button.clicked.connect(self.setup_thread)

    def setup_ui(self):
        self.setWindowTitle('demo')
        self.resize(QSize(250, 180))
        # 创建一个垂直布局
        layout = QVBoxLayout()
        # 创建一个标签
        self.label = QLabel('This is a label => ')
        layout.addWidget(self.label)
        # 创建一个按钮
        self.button = QPushButton('execute')
        layout.addWidget(self.button)
        # 将布局设置为主窗口的布局
        self.setLayout(layout)
        # 显示窗口
        self.show()

    def setup_thread(self):
        self.thread = MyThread(do_something)
        self.thread.finish.connect(self.thread_finished)
        self.thread.start()

    @Slot(str)
    def thread_finished(self, res):
        self.label.setText('This is a label => ' + res)


if __name__ == '__main__':
    app = QApplication([])
    window = MainWindow()
    window.show()
    app.exec()

线程池

代码释义

MyWorker类继承自QRunnable，接收一个函数、一个信号和其他可变参数。在run方法中，它调用了传入的函数并将结果发射出信号。

do_something函数是一个模拟的耗时操作，它在执行完毕后返回一个字符串。

MainWindow类是窗口的主要部分，它创建了一个垂直布局，并添加了一个标签和一个按钮。点击按钮后，它创建一个MyWorker对象，并使用线程池启动这个线程。同时，它将信号与thread_finished槽函数连接起来，在线程完成后更新标签的文本。

# -*- coding: utf-8 -*-

import time

from PySide6.QtCore import (QRunnable, QThreadPool, Signal, Slot, QSize)
from PySide6.QtWidgets import (QApplication, QPushButton, QLabel, QVBoxLayout, QWidget)


class MyWorker(QRunnable):

    def __init__(self, func, signal, *args, **kwargs):
        super().__init__()
        self.func = func
        self.args = args
        self.kwargs = kwargs
        self.signal = signal

    def run(self):
        res = self.func(*self.args, **self.kwargs)
        # 任务完成后发出信号
        self.signal.emit(res)


def do_something():
    time.sleep(1)
    return 'the function execution completed!'


class MainWindow(QWidget):
    signal = Signal(str)

    def __init__(self, parent=None):
        super().__init__(parent=parent)
        self.thread_pool = QThreadPool()
        self.setup_ui()
        self.button.clicked.connect(self.setup_thread)

    def setup_ui(self):
        self.setWindowTitle('demo')
        self.resize(QSize(250, 180))
        # 创建一个垂直布局
        layout = QVBoxLayout()
        # 创建一个标签
        self.label = QLabel('This is a label => ')
        layout.addWidget(self.label)
        # 创建一个按钮
        self.button = QPushButton('execute')
        layout.addWidget(self.button)
        # 将布局设置为主窗口的布局
        self.setLayout(layout)
        # 显示窗口
        self.show()

    def setup_thread(self):
        worker = MyWorker(do_something, self.signal)
        self.thread_pool.start(worker)
        self.signal.connect(self.thread_finished)

    @Slot(str)
    def thread_finished(self, res):
        self.label.setText('This is a label => ' + res)


if __name__ == '__main__':
    app = QApplication([])
    window = MainWindow()
    window.show()
    app.exec()