PyQt5 – 信号与槽机制详解

初次接触PyQt5的信号与槽机制,是在开发PyChat聊天软件时,需要实现后端服务接收消息的同时更新前端GUI界面,涉及到不同线程之间的对象通信。当初开发时走过不少弯路,现在回过头对这部分内容进行整理和记录。

部分内容整合自以下技术博客,深表感谢:
PyQt5快速入门(二)PyQt5信号槽机制 – https://blog.51cto.com/9291927/2422187
PyQt 5信号与槽的几种高级玩法 – https://blog.csdn.net/broadview2006/article/details/80132757

何谓信号与槽?

信号与槽机制适用于每一个QObject实例,它是实现GUI编程中对象之间通信的重要方法。信号与槽是相互匹配的关系,一个信号可以绑定多个槽,一个槽也可以监听多个信号。当某一信号被触发时,与之绑定的槽的函数会自动执行。这一机制类似于C/C++语言中的回调函数。回调函数通过把需要调用的函数指针传递给调用函数,来对特定事件做出响应。

用更通俗易懂的例子来说明,我们把信号与槽看做一种通知机制。比如我在电商上看中了一款商品,我想让电商平台在其降价时通知我。我们可以把信号看做是降价,槽看做通知我这一行为。我提交的“降价通知”请求,就是对信号与槽的绑定,而电商在降价时通知我这一过程则是槽接收到信号后所做的反应。

在PyQt5中,信号与槽机制具有如下特点:

  • 一个信号可以连接多个槽
  • 一个槽可以监听多个信号
  • 一个信号可以连接另一个信号
  • 信号参数可以是任何Python类型
  • 信号与槽的连接可以跨线程
  • 信号可以断开

定义信号与槽

在PyQt5中,我们使用pyqtsignal类来定义信号。该信号只能在QObject的子类中进行定义,且必须在创建类时进行定义,不可以进行动态添加。首先来看该类的__init__函数:

class pyqtSignal:
    def __init__(self, *types, name: str = ..., \
    revision:int = 0, arguments = []) -> None: ...

*types可以接受多个Python基本数据类型,表示该信号的参数类型;name接受自定义的信号名,默认为使用该信号类的属性名。例如下方代码中的notification对象。槽函数同样需要在QObject的子类中定义,可以是任意函数。revision与arguments参数在连接QML文件时使用,本文暂时不做探讨。定义后的signal将被自动添加到QObject的QMetaObject中。

from PyQt5.QtCore import *

class SignalClass(QObject):
    # 定义一个notification信号,接受一个list作为参数,信号名为printprice
    notification = pyqtSignal(int,name = "printprice")

    # 定义一个槽函数
    def send_pushmsg(self, price):
        print("Current Price:", price)

连接信号与槽 & 发送信号

在确定信号与槽后,需要对信号与槽进行绑定,以对特定事件做出反应。使用
QObject.signal.connect(slotFunction, type, bool: no_receiver_check)绑定信号与槽;type默认为自动连接方式,no_receiver_check为True时会忽视槽是否存在,强制发送信号;
使用QObject.singal.disconnect([slotFunctions])解绑信号与槽,可以同时解绑多个槽。使用signal.emit(*args) 来进行信号发射和参数传递。例如:

from PyQt5.QtCore import *

class SignalExample(QObject):
    # 定义一个notification信号,接受一个list作为参数,信号名为printprice
    notification = pyqtSignal(int,name = "printprice")

    # 定义一个槽函数
    def send_pushmsg(self, price):
        print("Current Price:", price)

signal = SignalExample()

# 将降价信号连接到推送消息函数上
signal.notification.connect(self.send_pushmsg)
# 此处也可以使用信号的别名
# signal.printprice.connect(self.send_pushmsg)

# 发射信号
signal.notification.emit(18)

Output: Current Price: 18

使用槽装饰器

槽装饰器通过装饰器的方法来定义信号和槽函数。在一般情况下,它会轻微降低内存占用,并轻微提升运行速度(由于在装饰器中直接声明了接收参数的类型,Python到C++接口的mapping会变得直接,而不需要程序去自动检测)。更为重要的一点是,槽装饰器允许对某一个槽进行重复的overload,并且声称不同的C++签名。

此外,在跨线程操作时,尤其涉及到以lambda函数作为槽时,需要注意PyQt会为lambda函数生成一个代理来满足信号/槽机制的需求,并且在新版的PyQt内(4.4+),信号与槽的连接方式直到信号被emit前才会确定。在连接lambda函数时,生成的代理会被相应地移动到接受对象所在的线程内:

    if (receive_qobj)
        proxy->moveToThread(receive_qobj->thread());

如果信号的接受对象(QObject)已经被移动到了正确的线程时,则不会产生问题,反之,可能会导致Proxy残留在主线程,导致无法正常工作。使用@pyqtSlot装饰器可以避免生成代理,直接生成一个相应的签名,从而避免这一问题。

PyQt5.QtCore.pyqtSlot(*types, name, result, revision=0)
*types接受传入的参数类型,name可以对槽进行重命名,result指定返回值的数据类型,revision同样用于导出槽函数到QML文件中,本文不做讨论。
具体的使用方法如下:

from PyQt5.QtCore import *

class SignalExample(QObject):
    # 定义一个notification信号,接受一个list作为参数,信号名为printlist
    notification = pyqtSignal(int,name = "printprice")

    # 定义一个槽函数
    # 此处,我们声明了接受和返回的参数类型
    @pyqtSlot(int, result=bool)
    def send_pushmsg(self, price):
        print("Current Price:", price)
        return True

signal = SignalExample()

# 将降价信号连接到推送消息函数上
signal.printprice.connect(signal.send_pushmsg)
# 发射信号
signal.notification.emit(18)

Output: Current Price: 18

断开信号与槽

直接使用PyQt5.QtCore.pyqtSignal.disconnect(pyqtSlot)即可。

实例:多线程中的信号与槽

本例中,我们将搭建前面提及到的价格监视器的最简易示例,使用两个线程,主进程为前台的GUI界面,一个线程作为后台的价格监控。主程序窗体上的价格将实时更新,当达到目标价格时将会进行提示。为方便测试,我们将初始价格设置为1000,目标价格设置为900。

from PyQt5.QtCore import *
from PyQt5.QtWidgets import *
import sys
import time


# 主程序窗体
class MainWindow(QWidget):
    def __init__(self, parent = None):
        super(MainWindow, self).__init__(parent)
        self.setWindowTitle("Price Monitor")
        self.resize(400, 200)

        # 创建价格文本框
        self.price = QLineEdit(self)
        self.price.setReadOnly(True)
        self.price.resize(400, 100)
        self.price.move(0, 20)

        # 创建退出按钮
        self.exitbutton = QPushButton("Exit",self)
        self.exitbutton.resize(self.exitbutton.sizeHint())
        self.exitbutton.move(135, 150)

        # 将退出按钮连接到QApplication的退出动作
        self.exitbutton.clicked.connect(QCoreApplication.instance().quit)
        self.loadUI()

    # 创建终止信号,用于控制终止后台监控线程
    terminal_sig = pyqtSignal()

    @pyqtSlot(str) # 更新价格信号槽,用于更新GUI的价格显示
    def update_price(self, price):
        self.price.setText(price)

    @pyqtSlot() # 通知信号槽,用于发出“达到目标价格”通知,并结束监控线程
    def notification(self):
        self.moniterThread.quit()
        self.price.setText(self.price.text() + "Reached Target Price!")
        # 向监控线程发出终止信号
        self.terminal_sig.emit()

    def loadUI(self):
        # 创建价格监控线程
        self.moniterThread = MonitorThread(1000, 900)
        # 连接监控线程的槽函数
        self.moniterThread.update_price.connect(self.update_price)
        self.moniterThread.notification.connect(self.notification)
        self.terminal_sig.connect(self.moniterThread.terminate)
        # 启动监控线程
        self.moniterThread.start()

# 价格监控线程
class MonitorThread(QThread):
    
    # 声明两个信号,更新价格信号与提醒信号
    update_price = pyqtSignal(str)
    notification = pyqtSignal()

    def __init__(self, initPrice, targetPrice):
        super().__init__()
        self.init_price = initPrice
        self.target_price = targetPrice

    def run(self):
        while True:
            # 为方便测试,价格每0.5秒减少10
            self.init_price -= 10
            # 向主进程发送通知,更新价格
            self.update_price.emit(str(self.init_price))
            if self.init_price == self.target_price:
                # 当达到目标价格时,向主进程发送通知
                self.notification.emit()
            time.sleep(0.5)

if __name__ == "__main__":
    app = QApplication(sys.argv)
    monitor = MainWindow()
    monitor.show()
    sys.exit(app.exec_())

当达到目标价格时,效果如图: