使用线程装饰器后如何检查线程是否存活？PyQt6开发问题

问题描述 #

我正在使用PyQt6开发GUI（若该框架不合适，欢迎推荐其他框架）。点击Connect按钮时，我需要验证连接线程未处于运行状态。我在主类的__init__中创建了self.master = ec.ecMaster()，尝试检查被@threaded装饰的ecConnection方法对应的线程是否存活，调用is_alive时出现错误：AttributeError: 'function' object has no attribute 'is_alive'。

现有代码 #

Main模块 #

from threading import thread
import pysoem
import ecAPI as ec
from PyQt6.QtWidgets import *
from PyQt6.QtCore import *
from PyQt6.QtGui import *
import logging
import sys
import time

class mainWindow(QMainWindow):
    exitSignal = Signal()  # 定义自定义退出信号

    def __init__(self,*args, **kwargs):        
        super().__init__(*args, **kwargs)
        self.exitSignal.connect(app.quit)
        self.central_widget = QWidget(self)
        self.test_widget = QWidget(self)
        self.modeSignal = ''
        self.operatorSignal = ''
        self.setCentralWidget(self.central_widget)
        self.connectionWindowLayout()
        self.master = ec.ecMaster()

    def connectButtonClicked(self):
        self.connect_button.clickable = False
        # 此处出现错误
        if self.master.ecConnection.is_alive:              # 检查线程是否活跃
            self.master.ecConnection.terminate()
            logger.info('连接线程已处于活跃状态，已终止')
        logger.info('启动连接线程')
        self.master.ecConnection()                      # 启动连接线程
        time.sleep(0.1)  

if __name__ == '__main__':
    
    logger = logging.getLogger(__name__)
    logging.getLogger().setLevel(logging.INFO)
    
    app = QApplication(sys.argv)
    # ModernStyle.apply(app)  # 自定义样式保留原代码
    window = mainWindow()
    window.show()
    sys.exit(app.exec())

ecAPI.py模块 #

import pysoem, logging
from threading import Thread

def threaded(fn):
    # 用作装饰器将函数调用转为线程执行
    def wrapper(*args, **kwargs):
        thread = Thread(target=fn, args=args, kwargs=kwargs)
        thread.start()
        return thread
    return wrapper

class ecMaster(pysoem.Master):
    def __init__(self):
        self.connectionSignal = 0
        super().__init__()
    
    def findAdapters(self):
        nicName = 'PCIe'            # 根据PC修改，示例为'PCIe'
        network_adapters = pysoem.find_adapters()
        for adapter in network_adapters:
            if nicName in str(adapter[1]):
                networkAdapterID = adapter[0]
                self.open(networkAdapterID)
                logging.info('找到网络适配器\n'+str(networkAdapterID))
        return len(network_adapters)>0 # 找到适配器返回True

    @threaded
    def ecConnection(self):
        if self.findAdapters():
            print('搜索EC设备中... ')
            self.connectionSignal = self.config_init()         # 初始化连接
            if self.slaves:
                print(f'找到 {len(self.slave_list)} 台设备')
                for i,slave in enumerate(self.slaves):
                    # ven = slave.sdo_read(0x1018, 1)
                    print(f'找到设备 {slave.name},{slave.id},{slave.man}') #,{slave.sdo_read()}
                    logging.info(f'找到 {self.connectionSignal} 台EC设备')
            else: 
                logging.info('未找到从站设备')
                print('未找到从站设备')
        else:
            logging.warning('未找到网络适配器')
            print('未找到网络适配器')

问题根源 #

你使用的@threaded装饰器每次调用ecConnection()时都会创建并返回一个新的Thread实例，但你没有保存这个实例，反而直接去访问原函数ecConnection的is_alive属性——原函数本身是普通函数对象，不是Thread实例，自然没有这个方法。

解决方案 #

1. 修改装饰器与ecMaster类，保存线程实例 #

修改ecAPI.py，给ecMaster类添加属性存储线程实例，并在装饰器中绑定这个实例：

import pysoem, logging
from threading import Thread

def threaded(fn):
    def wrapper(*args, **kwargs):
        # args[0]是ecMaster类的实例本身
        self = args[0]
        # 先检查是否已有活跃线程
        if hasattr(self, 'connection_thread') and self.connection_thread.is_alive():
            logging.info('连接线程已在运行，跳过启动')
            return self.connection_thread
        # 创建并启动新线程
        thread = Thread(target=fn, args=args, kwargs=kwargs)
        thread.start()
        # 将线程实例绑定到ecMaster对象
        self.connection_thread = thread
        return thread
    return wrapper

class ecMaster(pysoem.Master):
    def __init__(self):
        self.connectionSignal = 0
        self.connection_thread = None  # 初始化线程实例变量
        super().__init__()
    
    # 其余findAdapters、ecConnection方法保持不变

2. 修正主窗口的连接按钮逻辑 #

现在可以直接通过self.master.connection_thread检查线程状态：

def connectButtonClicked(self):
    self.connect_button.setEnabled(False)  # PyQt中用setEnabled控制按钮状态，不是clickable
    # 检查线程是否存在且处于活跃状态
    if self.master.connection_thread is not None and self.master.connection_thread.is_alive():
        # 先尝试优雅等待线程结束
        self.master.connection_thread.join(timeout=1)
        # 如果超时仍存活，再强制终止
        if self.master.connection_thread.is_alive():
            self.master.connection_thread.terminate()
        logger.info('连接线程已处于活跃状态，已终止')
    # 启动新线程
    logger.info('启动连接线程')
    self.master.ecConnection()
    # 不建议用time.sleep，会阻塞GUI，如需延迟操作改用QTimer
    # QTimer.singleShot(100, lambda: self.connect_button.setEnabled(True))
    self.connect_button.setEnabled(True)

额外建议 #

• PyQt6是非常合适的GUI框架，无需更换，若需要更美观的UI，可以搭配QDarkStyle或自定义QSS样式。
• 避免在PyQt主线程中使用time.sleep()，会导致界面卡顿，改用QTimer.singleShot()处理延迟逻辑。
• 如果ecConnection是一次性任务，可以在方法末尾添加self.connection_thread = None，避免后续误判线程状态。

内容的提问来源于stack exchange，提问作者Dizzy Dwarf