结合 Qt 信号槽机制的 Python 自定义线程类

pip install pyside2

# 导入可能用到的标准库模块
import <lib>

# 导入 PySide2 库模块
from PySide2.QtWidgets import QWidget
from PySide2.QtCore import Slot

# 导入其他自定义的模块包
import <custom_modules>
from <custom_modules> import <custom_class_or_function>


class MainWindow(QWidget):  
    """ MainWindow 继承 QWidget 类 """    
    def __init__(self):    
        QWidget.__init__(self)        
    
        # 设置类属性    
        pass      
    
        # 设置窗口属性    
        pass      
    
        # 创建窗口内部控件，并设置样式    
        self._create_components()    
        self._set_styles()      
    
    def _create_components(self):    
        """ 创建窗口内部的子控件 """    
        pass
    
    def _set_styles(self):    
        """ 设置窗口及其子控件的样式 """    
        pass
    
    @Slot()  
    def _slot_xxx_www(self):    
        """ 槽函数：连接至xxx控件的www信号 """    
        pass
    
    def _other_protected_function(self):    
        """ 其他的 protected 访问权限的方法 """    
        pass
    
    def other_public_function(self):    
        """ 其他的 public 访问权限的方法 """    
        pass

import sys
from PySide2.QtWidgets import QApplication
from main_windows import MainWindow


if __name__ == "__main__":  
    app = QApplication(sys.argv)     # 创建 Qt 应用程序实例  
    main_win = MainWindow()          # 创建主窗口实例  
    main_win.show()                  # 显示主窗口  
    sys.exit(app.exec_())            # 进入循环监听事件

from PySide2.QtWidgets import QPushButton
from PySide2.QtCore import Slot


@Slot()
def slot_function():  
    """  定义槽函数：处理按钮被单击时的事件  
    其中 @Slot() 为槽函数的装饰器，其实也可以不写装饰器，但为了便于区分普通的函数，还是建议加上装饰器。  
    """  
    pass


# 实例化一个 Qt 控件（如按钮）
btn = QPushButton()

# 直接将控件的某个信号（如按钮的 clicked 信号）使用 connect 方法连接到槽函数
btn.clicked.connect(slot_function)

from PySide2.QtWidgets import QApplication
from PySide2.QtCore import Qt, QObject, Signal, Slot


class CustomClass(QObject):  
    """ 这是一个自定义的类，在这个类里面要使用 Qt 的信号，因此必须继承 QObject 类 """    
    
    # 定义一个信号，注意信号的定义只能写在类属性中，不能作为对象属性定义在构造方法中。  
    custom_signal1 = Signal()    
    
    # 定义一个带参数的信号，此时只需声明参数的类型即可  
    custom_signal2 = Signal(int, str)    
    
    def __init__(self):    
        QObject.__init__(self)      
        
    def send_signal(self):    
        """ 调用该方法时向外部发送信号 """    
        self.custom_signal1.emit()              # 使用 emit() 方法发射信号    
        self.custom_signal2.emit(1, "hello")    # 带参数的信号则需一起发送参数        
        
    @Slot()
    def slot_deal_signal1():  
        """    定义槽函数，用于处理 custom_signal1 信号  """  
        print("get custom signal 1.")    
        
    @Slot(int, str)
    def slot_deal_signal2(num: int, text: str):  
        """  定义槽函数：用于处理 custom_signal2 信号  
        注意此时装饰器以及函数的参数列表应该与信号中定义的参数列表一致  
        """  
        print("get custom signal 2.")  
        print("the first arg is:", num)  
        print("the second arg is:", text)    
        
        
# 实例化这个类
cs = CustomClass()

# 连接信号与槽
cs.custom_signal1.connect(slot_deal_signal1)
cs.custom_signal2.connect(slot_deal_signal2)

# 发射信号
cs.send_signal()

get custom signal 1.
get custom signal 2.
the first arg is: 1
the second arg is: hello

import time
from threading import Thread


class PrintTimeThread(Thread):  
    """ 自定义线程类：用于打印当前时间 """    
    
    def __init__(self):    
        Thread.__init__(self)      
        
    def run(self):    
        """ 重写 run 方法：实现子线程处理逻辑 """    
        while True:      
            print(time.asctime())      # 打印时间      
            time.sleep(1)              # 休眠1秒            
            

if __name__ == "__main__":  
    # 实例化子线程对象  
    my_thread = PrintTimeThread()    
    
    # 启动子线程  
    my_thread.start()    
    
    # 为了作对比，主线程中每隔三秒打印一次 “hello”  
    while True:    
        print("hello")    
        time.sleep(3)

import time
from threading import Thread
from PySide2.QtCore import QObject, Signal, Slot


class GetTimeThread(Thread, QObject):  
    """   自定义线程类：用于打印当前时间   
    注意这里使用 Python 多继承的概念，同时继承了 Thread 类和 QObjet 类  
    """    
    
    # 定义信号：每隔一秒钟触发信号，并将当前时间信息发射出去  
    # 注意这里使用了带参数的信号，待发射的时间信息为 str 类型  
    clock_signal = Signal(str)    
    
    def __init__(self):    
        # 父类初始化    
        Thread.__init__(self)    
        QObject.__init__(self)      
        
    def run(self):    
        """ 重写 run 方法：实现子线程处理逻辑 """    
        while True:      
            # print(time.asctime())                  # 现在不用在子线程中打印时间了，所以注释掉      
            self.clock_signal.emit(time.asctime())   # 直接将时间信息发射出去      
            time.sleep(1)                            # 休眠1秒            
            
    @Slot(str)
    def slot_print_time(current_time: str):  
        """ 定义槽函数：接收子线程传送过来的时间信息，并打印出来   
        :param current_time: 由子线程发射过来的时间信息  
        """  
        print(current_time)            
        
        
if __name__ == "__main__":  
    # 实例化子线程对象  
    my_thread = PrintTimeThread()    
    
    # 连接子线程的信号到主线程的槽函数中  
    my_thread.clock_signal.connect(slot_print_time)    
    
    # 启动子线程  
    my_thread.start()    
    
    # 为了作对比，主线程中每隔三秒打印一次 “hello”  
    while True:    
        print("hello")    
        time.sleep(3)