QThread 多执行绪

使用 PyQt5 设计界面时，视窗主程序的本质是放在一个“无穷循环”里执行，如果需要加入多个循环且不影响主视窗 ( 如果单纯放入循环，会在所有循环结束后才启动视窗 )，就需要使用 QThread 机制，让多个执行绪同时执行，这篇教学会介绍 QThread 的使用方式，还会额外介绍搭配 Python threading 标准函数库的作法。

快速导览：

• 开始使用 QThread
• QThread 常用方法
• QThread 搭配 threading Event
• 使用 threading + PyQt5

• PyQt6 版本参考：PyQt6 教学 - QThread 多执行绪
• 因为 Google Colab 不支援 PyQt5，所以请使用本机环境 ( 参考：使用 Python 虚拟环境 ) 或使用 Anaconda Jupyter 进行实作 ( 参考：使用 Anaconda )。

开始使用 QThread

建立 PyQt5 视窗对象后，如果要执行的程序内容里出现了“循环”，则因为程序码执行的顺序，必须要等待所有循环结束后，才会正式启动视窗，例如下方的程序码，视窗里放入两个 QLabel，分别使用 a 和 b 两个函数，透过循环的方式设定文字内容不断改变，但因为里面用到 time.sleep 的方法主视窗就会等待所有循环结束后才会启动，造成视窗出现后，QLabel 里的文字已经固定为最终的结果 ( 如果观察 print 的内容，也会发现 a() 执行完毕后才会执行 b() )。

from PyQt5 import QtWidgets
import sys, time

app = QtWidgets.QApplication(sys.argv)

Form = QtWidgets.QWidget()
Form.setWindowTitle('oxxo.studio')
Form.resize(300, 200)

label_a = QtWidgets.QLabel(Form)      # 第一個 QLabel
label_a.setGeometry(10, 10, 100, 30)

label_b = QtWidgets.QLabel(Form)      # 第二個 QLabel
label_b.setGeometry(10, 50, 100, 30)

def a():
    for i in range(0,5):
        label_a.setText(str(i))      # 每次迴圈執行時設定文字
        print('A:',i)
        time.sleep(0.5)              # 等待 0.5 秒

def b():
    for i in range(0,50,10):
        label_b.setText(str(i))      # 每次迴圈執行時設定文字
        print('B:',i)
        time.sleep(0.5)              # 等待 0.5 秒

a()          # 執行 a()
b()          # 執行 a()

Form.show()  # 顯示主視窗
sys.exit(app.exec_())

如果要让程序正常运作，除了使用 PyQt5 的 QTimer 定时器，也可透过 QThread 多执行绪的方式，分别将函数放入不同的执行绪中去执行，就能做到“异步”的效果 ( 参考“同步与异步”)，下方的程序码先 import QThread 模组，接着定义出两个 QThread 执行绪，再将执行绪的 run 对应到要执行的函数，执行后就会是预期的结果。

from PyQt5 import QtWidgets
from PyQt5.QtCore import QThread
import sys, time

app = QtWidgets.QApplication(sys.argv)

Form = QtWidgets.QWidget()
Form.setWindowTitle('oxxo.studio')
Form.resize(300, 200)

label_a = QtWidgets.QLabel(Form)
label_a.setGeometry(10, 10, 100, 30)

label_b = QtWidgets.QLabel(Form)
label_b.setGeometry(10, 50, 100, 30)

def a():
    for i in range(0,5):
        label_a.setText(str(i))
        print('A:',i)
        time.sleep(0.5)

def b():
    for i in range(0,50,10):
        label_b.setText(str(i))
        print('B:',i)
        time.sleep(0.5)

thread_a = QThread()   # 建立 Thread()
thread_a.run = a       # 設定該執行緒執行 a()
thread_a.start()       # 啟動執行緒

thread_b = QThread()   # 建立 Thread()
thread_b.run = b       # 設定該執行緒執行 b()
thread_b.start()       # 啟動執行緒

Form.show()
sys.exit(app.exec_())

QThread 常用方法

下列 QThread 常用方法，可以处理执行绪的开始与等待：

|方法|参数|说明| |--|--| |start()||启动执行绪。| |wait()||等待该执行绪结束。| |sleep()|sec|等待该执行绪几秒。|

举例来说，修改上面的程序码，在 b 函数里加入 thread_a.wait()，b 函数的循环就会等到 a 函数执行完成后 ( thread_a 执行绪结束 )，才会开始动作。

def b():
    thread_a.wait()             # 使用 wait
    for i in range(0,50,10):
        label_b.setText(str(i))
        print('B:',i)
        time.sleep(0.5)

如果将 thread_a.wait() 改成 thread_a.sleep(1)，则 b 函数的循环就会等到 a 函数执行一秒后，再开始动作。

def b():
    thread_a.sleep(1)             # 使用 sleep
    for i in range(0,50,10):
        label_b.setText(str(i))
        print('B:',i)
        time.sleep(0.5)

QThread 搭配 threading Event

搭配 threading 标准函数库的 Event 功能 ( 需要 import threading )，透过侦测事件的方式，就能要让不同的 QThread 之间可以互相沟通连动 ( 例如 a 执行到发生某件事，再让 b 开始动作 )，下方的程序码执行后，会在 b 函数进行数字大于 20 的时候触发事件，这时 a 函数收到事件被触发的消息，就会开始动作。

from PyQt5 import QtWidgets
from PyQt5.QtCore import QThread
import sys, time, threading

app = QtWidgets.QApplication(sys.argv)

Form = QtWidgets.QWidget()
Form.setWindowTitle('oxxo.studio')
Form.resize(300, 200)

label_a = QtWidgets.QLabel(Form)
label_a.setGeometry(10, 10, 100, 30)

label_b = QtWidgets.QLabel(Form)
label_b.setGeometry(10, 50, 100, 30)

event = threading.Event()    # 建立事件

def a():
    event.wait()             # 等待事件被觸發
    for i in range(0,5):
        label_a.setText(str(i))
        print('A:',i)
        time.sleep(0.5)

def b():
    for i in range(0,50,10):
        if i>20:
            event.set()      # 觸發事件
        label_b.setText(str(i))
        print('B:',i)
        time.sleep(0.5)

thread_a = QThread()
thread_a.run = a
thread_a.start()

thread_b = QThread()
thread_b.run = b
thread_b.start()

Form.show()
sys.exit(app.exec_())

上方的程序码也可以改成用 class 的写法：

from PyQt5 import QtWidgets
from PyQt5.QtCore import QThread
import sys, time, threading

class MyWidget(QtWidgets.QWidget):
    def __init__(self):
        super().__init__()
        self.setWindowTitle('oxxo.studio')
        self.resize(300, 200)
        self.event = threading.Event()
        self.ui()
        self.run()

    def ui(self):
        self.label_a = QtWidgets.QLabel(self)
        self.label_a.setGeometry(10, 10, 100, 30)

        self.label_b = QtWidgets.QLabel(self)
        self.label_b.setGeometry(10, 50, 100, 30)

    def a(self):
        self.event.wait()             # 等待事件被觸發
        for i in range(0,5):
            self.label_a.setText(str(i))
            print('A:',i)
            time.sleep(0.5)

    def b(self):
        for i in range(0,50,10):
            if i>20:
                self.event.set()      # 觸發事件
            self.label_b.setText(str(i))
            print('B:',i)
            time.sleep(0.5)

    def run(self):
        self.thread_a = QThread()
        self.thread_a.run = self.a
        self.thread_a.start()

        self.thread_b = QThread()
        self.thread_b.run = self.b
        self.thread_b.start()

if __name__ == '__main__':
    app = QtWidgets.QApplication(sys.argv)
    Form = MyWidget()
    Form.show()
    sys.exit(app.exec_())

使用 threading + PyQt5

除了使用 PyQt5 内建的 QThread，也可以单纯使用 Python 标准函数库 threading 来处理多执行绪 ( 参考“threading 多执行绪处理” )，下方的程序码使用 threading 的方式，执行后会产生跟使用 QThread 同样的效果。

from PyQt5 import QtWidgets
from PyQt5.QtCore import QThread
import sys, time, threading

app = QtWidgets.QApplication(sys.argv)

Form = QtWidgets.QWidget()
Form.setWindowTitle('oxxo.studio')
Form.resize(300, 200)

label_a = QtWidgets.QLabel(Form)
label_a.setGeometry(10, 10, 100, 30)

label_b = QtWidgets.QLabel(Form)
label_b.setGeometry(10, 50, 100, 30)

def a():
    for i in range(1,6):
        label_a.setText(str(i))
        print('A:',i)
        time.sleep(0.5)

def b():
    for i in range(10,60,10):
        label_b.setText(str(i))
        print('B:',i)
        time.sleep(0.5)

thread_a = threading.Thread(target=a)   # 建立執行緒，執行 a 函式
thread_b = threading.Thread(target=b)   # 建立執行緒，執行 b 函式

thread_a.start()   # 啟動執行緒
thread_b.start()   # 啟動執行緒

Form.show()
sys.exit(app.exec_())

意见反馈