PySide6 GUI 编程（44）：异步执行 QRunnable 与 QThreadPool

一个简单的示例

示例代码

代码语言：python代码运行次数：0复制Cloud Studio 代码运行

from __future__ import annotations

import sys
import time
from datetime import datetime

from PySide6.QtCore import Qt, QTimer
from PySide6.QtGui import QFont
from PySide6.QtWidgets import QApplication, QLabel, QMainWindow, QPushButton, QVBoxLayout, QWidget


def get_time_str() -> str:
    return datetime.now().isoformat(sep = ' ')


def sleep_block() -> None:
    """
    使用 time.sleep(interval) 来暂停程序的执行
    这会导致主线程（即 GUI 线程）被阻塞，无法处理任何其他事件（如更新界面、响应按钮点击等）
    因此，用户在点击按钮后，界面会冻结，直到 sleep 完
    """
    interval = 3
    print('{} begin sleep {}s'.format(get_time_str(), interval))
    time.sleep(interval)
    print('{} end sleep {}s'.format(get_time_str(), interval))


class MyMainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle('Hello, PySide6!')
        self.setToolTip('A PySide6 GUI Application Demo')
        self.setFixedSize(600, 400)

        # 设置计数器
        self.counter = 0
        # 设置定时器，每隔2秒执行一次update_label()方法
        self.timer = QTimer()
        self.timer.timeout.connect(self.update_label)
        self.timer.setInterval(1000)

        # 在窗口中添加标签
        self.label = QLabel(f'{get_time_str()} COUNTER: {self.counter}')
        self.label.setAlignment(Qt.AlignmentFlag.AlignCenter)
        self.label.setWordWrap(True)
        self.label.setFont(QFont('ComicShannsMono Nerd Font Propo', 25))
        self.label.setStyleSheet("color: green;")

        # 在窗口中添加按钮
        self.button = QPushButton('按下就异常')
        self.button.clicked.connect(sleep_block)

        # 布局
        self.v_layout = QVBoxLayout()
        self.v_layout.addWidget(self.label)
        self.v_layout.addWidget(self.button)
        container = QWidget()
        container.setLayout(self.v_layout)
        self.setCentralWidget(container)

        # 启动定时器
        self.timer.start()

    def update_label(self) -> None:
        self.counter  = 1
        self.label.setText(f'{get_time_str()} COUNTER: {self.counter}')


if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = MyMainWindow()
    window.show()
    app.exec()

示例中运行了一个定时器，每隔一秒中计数一次。

当按下按钮时，会触发 sleep_block 进入 sleep 逻辑，这个逻辑会导致主线程卡住，影响界面的交互。

我们可以把按钮按下去后触发的逻辑看作是一个耗时比较久的动作，比如下载资源的过程或者计算的过程，当我们触发了这样的耗时操作后，就会导致整个主界面被卡住。

示例效果

主线程阻塞

基于 QRunnable 和 QThreadPool 异步执行耗时逻辑

示例代码

代码语言：python代码运行次数：0复制Cloud Studio 代码运行

from __future__ import annotations

import sys
import threading
import time
from datetime import datetime

from PySide6.QtCore import QRunnable, Qt, QThreadPool, QTimer, Slot
from PySide6.QtGui import QFont
from PySide6.QtWidgets import QApplication, QLabel, QMainWindow, QPushButton, QVBoxLayout, QWidget


def get_time_str() -> str:
    return datetime.now().isoformat(sep = ' ')


def sleep_block() -> None:
    """
    使用 time.sleep(interval) 来暂停程序的执行
    这会导致主线程（即 GUI 线程）被阻塞，无法处理任何其他事件（如更新界面、响应按钮点击等）
    因此，用户在点击按钮后，界面会冻结，直到 sleep 完
    """
    interval = 3
    print('ID:{}, {} begin sleep {}s'.format(threading.get_ident(), get_time_str(), interval))
    time.sleep(interval)
    print('ID:{}, {} end sleep {}s'.format(threading.get_ident(), get_time_str(), interval))


class BlockWorker(QRunnable):
    def __init__(self):
        super().__init__()

    @Slot()
    def run(self) -> None:
        sleep_block()


class MyMainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle('Hello, PySide6!')
        self.setToolTip('A PySide6 GUI Application Demo')
        self.setFixedSize(600, 400)

        self.threads = QThreadPool()
        print('maximum threads: {}'.format(self.threads.maxThreadCount()))

        # 设置计数器
        self.counter = 0
        # 设置定时器，每隔2秒执行一次update_label()方法
        self.timer = QTimer()
        self.timer.timeout.connect(self.update_label)
        self.timer.setInterval(1000)

        # 在窗口中添加标签
        self.label = QLabel(f'{get_time_str()} COUNTER: {self.counter}nMAX threads count:{self.threads.maxThreadCount()}')
        self.label.setAlignment(Qt.AlignmentFlag.AlignCenter)
        self.label.setWordWrap(True)
        self.label.setFont(QFont('ComicShannsMono Nerd Font Propo', 25))
        self.label.setStyleSheet("color: green;")

        # 在窗口中添加按钮
        self.button = QPushButton('开启异步线程')
        self.button.clicked.connect(self.start_threads)

        # 布局
        self.v_layout = QVBoxLayout()
        self.v_layout.addWidget(self.label)
        self.v_layout.addWidget(self.button)
        container = QWidget()
        container.setLayout(self.v_layout)
        self.setCentralWidget(container)

        # 启动定时器
        self.timer.start()

    def update_label(self) -> None:
        self.counter  = 1
        self.label.setText(f'{get_time_str()} COUNTER: {self.counter}nMAX threads count:{self.threads.maxThreadCount()}')

    def start_threads(self) -> None:
        worker = BlockWorker()
        self.threads.start(worker)


if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = MyMainWindow()
    window.show()
    app.exec()

实现可异步执行的执行单元

开启线程

示例效果

开启异步线程

QThreadPool 可以对任意槽函数开启线程

示例代码

代码语言：python代码运行次数：0复制Cloud Studio 代码运行

from __future__ import annotations

import sys
import threading
import time
from datetime import datetime

from PySide6.QtCore import Qt, QThreadPool, QTimer, Slot
from PySide6.QtGui import QFont
from PySide6.QtWidgets import QApplication, QLabel, QMainWindow, QPushButton, QVBoxLayout, QWidget


def get_time_str() -> str:
    return datetime.now().isoformat(sep = ' ')


class MyMainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle('Hello, PySide6!')
        self.setToolTip('A PySide6 GUI Application Demo')
        self.setFixedSize(600, 400)

        self.threads = QThreadPool()
        print('maximum threads: {}'.format(self.threads.maxThreadCount()))

        # 设置计数器
        self.counter = 0
        # 设置定时器，每隔2秒执行一次update_label()方法
        self.timer = QTimer()
        self.timer.timeout.connect(self.update_label)
        self.timer.setInterval(1000)

        # 在窗口中添加标签
        self.label = QLabel(f'{get_time_str()} COUNTER: {self.counter}nMAX threads count:{self.threads.maxThreadCount()}')
        self.label.setAlignment(Qt.AlignmentFlag.AlignCenter)
        self.label.setWordWrap(True)
        self.label.setFont(QFont('ComicShannsMono Nerd Font Propo', 25))
        self.label.setStyleSheet("color: green;")

        # 在窗口中添加按钮
        self.button = QPushButton('开启异步线程')
        self.button.clicked.connect(self.start_threads)

        # 布局
        self.v_layout = QVBoxLayout()
        self.v_layout.addWidget(self.label)
        self.v_layout.addWidget(self.button)
        container = QWidget()
        container.setLayout(self.v_layout)
        self.setCentralWidget(container)

        # 启动定时器
        self.timer.start()

    def update_label(self) -> None:
        self.counter  = 1
        self.label.setText(f'{get_time_str()} COUNTER: {self.counter}nMAX threads count:{self.threads.maxThreadCount()}')

    @Slot()
    def sleep_block(self) -> None:
        """
        for simple use-cases,
        Qt provides a convenience method through QThreadPool.start() which can handle the execution of
        arbitrary Python functions and methods.
        Qt creates the necessary QRunnable objects for you and queues them on the pool.

        QThreadPool.start() 方法可以处理任意的QMainWindow槽函数
        """
        interval = 3
        print('ID:{}, {} begin sleep {}s'.format(threading.get_ident(), get_time_str(), interval))
        time.sleep(interval)
        print('ID:{}, {} end sleep {}s'.format(threading.get_ident(), get_time_str(), interval))

    def start_threads(self) -> None:
        self.threads.start(self.sleep_block)


if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = MyMainWindow()
    window.show()
    app.exec()

直接对槽函数开启异步线程

示例效果

对槽函数开启异步线程

QThreadPool 对非 QMainWindow 类槽函数生效

示例代码

代码语言：python代码运行次数：0复制Cloud Studio 代码运行

from __future__ import annotations

import sys
import threading
import time
from datetime import datetime

from PySide6.QtCore import Qt, QThreadPool, QTimer, Slot
from PySide6.QtGui import QFont
from PySide6.QtWidgets import QApplication, QLabel, QMainWindow, QPushButton, QVBoxLayout, QWidget


def get_time_str() -> str:
    return datetime.now().isoformat(sep = ' ')


@Slot()
def sleep_block() -> None:
    """
    如果 sleep_block 函数是一个独立的函数，而不是 MyMainWindow 类的成员方法
    则会导致无法在 start_threads 方法中直接调用 sleep_block
    """
    interval = 3
    print('ID:{}, {} begin sleep {}s'.format(threading.get_ident(), get_time_str(), interval))
    time.sleep(interval)
    print('ID:{}, {} end sleep {}s'.format(threading.get_ident(), get_time_str(), interval))


class MyMainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle('Hello, PySide6!')
        self.setToolTip('A PySide6 GUI Application Demo')
        self.setFixedSize(600, 400)

        self.threads = QThreadPool()
        print('maximum threads: {}'.format(self.threads.maxThreadCount()))

        # 设置计数器
        self.counter = 0
        # 设置定时器，每隔2秒执行一次update_label()方法
        self.timer = QTimer()
        self.timer.timeout.connect(self.update_label)
        self.timer.setInterval(1000)

        # 在窗口中添加标签
        self.label = QLabel(f'{get_time_str()} COUNTER: {self.counter}nMAX threads count:{self.threads.maxThreadCount()}')
        self.label.setAlignment(Qt.AlignmentFlag.AlignCenter)
        self.label.setWordWrap(True)
        self.label.setFont(QFont('ComicShannsMono Nerd Font Propo', 25))
        self.label.setStyleSheet("color: green;")

        # 在窗口中添加按钮
        self.button = QPushButton('开启异步线程')
        self.button.clicked.connect(self.start_threads)

        # 布局
        self.v_layout = QVBoxLayout()
        self.v_layout.addWidget(self.label)
        self.v_layout.addWidget(self.button)
        container = QWidget()
        container.setLayout(self.v_layout)
        self.setCentralWidget(container)

        # 启动定时器
        self.timer.start()

    def update_label(self) -> None:
        self.counter  = 1
        self.label.setText(f'{get_time_str()} COUNTER: {self.counter}nMAX threads count:{self.threads.maxThreadCount()}')

    def start_threads(self) -> None:
        self.threads.start(sleep_block)


if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = MyMainWindow()
    window.show()
    app.exec()

非 QMainWindow 类槽函数

示例效果

非 QMainWindow 类槽函数正常运行

为QRunnable类传递参数

示例代码

代码语言：python代码运行次数：0复制Cloud Studio 代码运行

from __future__ import annotations

import sys
import threading
import time
from datetime import datetime
from random import randint

from PySide6.QtCore import QRunnable, Qt, QThreadPool, QTimer, Slot
from PySide6.QtGui import QFont
from PySide6.QtWidgets import QApplication, QLabel, QMainWindow, QPushButton, QVBoxLayout, QWidget


def get_time_str() -> str:
    return datetime.now().isoformat(sep = ' ')


class MyWorker(QRunnable):
    def __init__(self, *args, **kwargs):
        # 传递参数给 Worker
        super().__init__()
        print('args=', args, 'kwargs=', kwargs)

    @Slot()
    def run(self) -> None:
        print('ID:{}, {} begin sleep 1s'.format(threading.get_ident(), get_time_str()))
        time.sleep(1)
        print('ID:{}, {} end sleep 1s'.format(threading.get_ident(), get_time_str()))


class MyMainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle('Hello, PySide6!')
        self.setToolTip('A PySide6 GUI Application Demo')
        self.setFixedSize(600, 400)

        self.threads = QThreadPool()
        print('maximum threads: {}'.format(self.threads.maxThreadCount()))

        # 设置计数器
        self.counter = 0
        # 设置定时器，每隔2秒执行一次update_label()方法
        self.timer = QTimer()
        self.timer.timeout.connect(self.update_label)
        self.timer.setInterval(1000)

        # 在窗口中添加标签
        self.label = QLabel(f'{get_time_str()} COUNTER: {self.counter}nMAX threads count:{self.threads.maxThreadCount()}')
        self.label.setAlignment(Qt.AlignmentFlag.AlignCenter)
        self.label.setWordWrap(True)
        self.label.setFont(QFont('ComicShannsMono Nerd Font Propo', 25))
        self.label.setStyleSheet("color: green;")

        # 在窗口中添加按钮
        self.button = QPushButton('开启异步线程')
        self.button.clicked.connect(self.start_threads)

        # 布局
        self.v_layout = QVBoxLayout()
        self.v_layout.addWidget(self.label)
        self.v_layout.addWidget(self.button)
        container = QWidget()
        container.setLayout(self.v_layout)
        self.setCentralWidget(container)

        # 启动定时器
        self.timer.start()

    def update_label(self) -> None:
        self.counter  = 1
        self.label.setText(f'{get_time_str()} COUNTER: {self.counter}nMAX threads count:{self.threads.maxThreadCount()}')

    def start_threads(self) -> None:
        self.threads.start(
            MyWorker([randint(199, 599) for _ in range(5)], arg1 = 'hello', arg2 = 'pyside6', arg3 = randint(1000, 9999)))


if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = MyMainWindow()
    window.show()
    app.exec()

QRunnable 初始化参数

实例化时传入参数

示例效果

实例化时传入参数

QRunnable发出信号

示例代码

代码语言：python代码运行次数：0复制Cloud Studio 代码运行

from __future__ import annotations

import sys
import threading
import time
from datetime import datetime
from random import randint

from PySide6.QtCore import QObject, QRunnable, Qt, QThreadPool, QTimer, Signal, Slot
from PySide6.QtGui import QFont
from PySide6.QtWidgets import QApplication, QLabel, QMainWindow, QPushButton, QVBoxLayout, QWidget


def get_time_str() -> str:
    return datetime.now().isoformat(sep = ' ')


class WorkerSignals(QObject):
    begin = Signal(int, str, int)
    finished = Signal(str, str, tuple, dict)


class MyWorker(QRunnable):
    def __init__(self, *args, **kwargs):
        # 传递参数给 Worker
        super().__init__()
        # print('args=', args, 'kwargs=', kwargs)
        # 初始化信号和槽
        self.signals = WorkerSignals()
        self.args = args
        self.kwargs = kwargs

    @Slot()
    def run(self) -> None:
        interval = randint(1, 5)
        # threading.get_ident() 返回的值可能会超过 pyside6中整数的上限值
        # RuntimeWarning: libshiboken: Overflow: Value 5 exceeds limits of type  [signed] "x" (8bytes).
        #   self.signals.finished.emit(threading.get_ident(),
        # 因此这里将 int 转换为 string
        id = threading.get_ident()
        self.signals.begin.emit(str(id),
                                get_time_str(),
                                interval)
        time.sleep(interval)
        self.signals.finished.emit(str(id),
                                   get_time_str(),
                                   self.args,
                                   self.kwargs)


def handle_worker_begin(thread_id: str, time_str: str, interval: int) -> None:
    print('BEGIN thread_id=', thread_id, ', time_str=', time_str, ', sleep {}s'.format(interval))


def handle_worker_finished(thread_id: str, time_str: str, args: tuple, kwargs: dict) -> None:
    print('FINISHED thread_id=', thread_id, ', time_str=', time_str, ', args=', args, ', kwargs=', kwargs)


class MyMainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle('Hello, PySide6!')
        self.setToolTip('A PySide6 GUI Application Demo')
        self.setFixedSize(600, 400)

        self.threads = QThreadPool()
        print('maximum threads: {}'.format(self.threads.maxThreadCount()))

        # 设置计数器
        self.counter = 0
        # 设置定时器，每隔2秒执行一次update_label()方法
        self.timer = QTimer()
        self.timer.timeout.connect(self.update_label)
        self.timer.setInterval(1000)

        # 在窗口中添加标签
        self.label = QLabel(f'{get_time_str()} COUNTER: {self.counter}nMAX threads count:{self.threads.maxThreadCount()}')
        self.label.setAlignment(Qt.AlignmentFlag.AlignCenter)
        self.label.setWordWrap(True)
        self.label.setFont(QFont('ComicShannsMono Nerd Font Propo', 25))
        self.label.setStyleSheet("color: green;")

        # 在窗口中添加按钮
        self.button = QPushButton('开启异步线程')
        self.button.clicked.connect(self.start_threads)

        # 布局
        self.v_layout = QVBoxLayout()
        self.v_layout.addWidget(self.label)
        self.v_layout.addWidget(self.button)
        container = QWidget()
        container.setLayout(self.v_layout)
        self.setCentralWidget(container)

        # 启动定时器
        self.timer.start()

    def update_label(self) -> None:
        self.counter  = 1
        self.label.setText(f'{get_time_str()} COUNTER: {self.counter}nMAX threads count:{self.threads.maxThreadCount()}')

    def start_threads(self) -> None:
        worker = MyWorker([randint(9999, 999999) for _ in range(10)],
                          arg1 = 'hello',
                          arg2 = 'world',
                          arg3 = {'a': '1', 'b': 2, 'c': [1, 2, 3, 4, 5, 6]},
                          )
        worker.signals.begin.connect(handle_worker_begin)
        worker.signals.finished.connect(handle_worker_finished)
        self.threads.start(worker)


if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = MyMainWindow()
    window.show()
    app.exec()

QRunnable 实例发出信号

绑定槽函数

示例效果

QRunnable 实例通过信号传出数据

特别注意

整数溢出问题

PySide6 中 int 存在溢出问题，需要转换为 str

int 转 str 可以规避溢出问题

pythonPySide6 GUI编程 Python3

0 人点赞