You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python Worker线程类偶发异常及Tk askopenfilename主线程报错求助

线程Worker脚本偶发异常及Tk弹窗调用错误问题

我编写了一个名为worker的Python脚本,包含Worker和CreateThread两个类,用于处理意外异常并通过线程运行耗时函数以避免GUI冻结。但该脚本存在偶发异常:有时调用Worker类运行函数一切正常,有时却出现奇怪错误甚至GUI崩溃。我尝试调整线程处理顺序、适配错误,但仍存在隐藏问题未解决。

以下是我的worker代码:

# Create a worker class #
class Worker(QObject):
    my_started = pyqtSignal()
    my_finished = pyqtSignal()
    my_error = pyqtSignal(str, str)
    my_success = pyqtSignal()

    def __init__(self):
        super(Worker, self).__init__()

    def run(self, func):
        try:
            self.my_started.emit()
            time.sleep(0.1)
            x = func()
            if x is None:
                self.my_success.emit()
                self.my_finished.emit()
                return

            if x[0] == -1:
                function_error = x[1]
                trace_back = x[2]
                print(f'\nFunction error: {function_error}')
                print(trace_back)
                self.my_error.emit(str(function_error), trace_back,
                                   str(func.__name__))
                self.my_finished.emit()
                return

        except Exception as e:
            self.trace_back = str(traceback.format_exc())
            print(f'\nFunction error: {e}')
            print(self.trace_back)
            self.my_error.emit(str(e), self.trace_back)

        self.my_finished.emit()


# Creating thread manger class #
class CreateThread(Worker, QObject):

    def __init__(self, atp_name):
        super(CreateThread, self).__init__()
        self.stop_flag = False
        self.atp_name = atp_name
        self.thread = QThread()
        self.worker = Worker()

    def start_thread(self, func, **kwargs):
        self.kwargs = kwargs

        self.func_name = func.__name__

        if self.stop_flag:
            return

        if self.func_name == '<lambda>':
            self.func_name = self.kwargs['func_name']

        # Thread and Worker init #
        self.thread = QThread()
        self.worker = Worker()
        act = partial(self.worker.run, func)
        self.worker.moveToThread(self.thread)
        self.thread.started.connect(act)

        # Signal to Function connection #
        self.worker.my_started.connect(self.started_signal_connection)
        self.worker.my_finished.connect(self.finished_signal_connection)
        self.worker.my_error.connect(self.error_signal_connection)
        self.worker.my_success.connect(self.success_signal_connection)

        # Handling the worker and the thread deletion #
        self.worker.my_finished.connect(self.thread.quit)
        self.worker.my_finished.connect(self.worker.deleteLater)
        self.thread.finished.connect(self.thread.deleteLater)

        self.thread.start()

    # Connect signals from worker #
    def started_signal_connection(self):
        self.stop_flag = True
        print(f"\nFunction {self.func_name} started\n")

    def finished_signal_connection(self):
        self.stop_flag = False
        print("\nWorker finished\n")

    def error_signal_connection(self, error_name, trace_back):
        self.today = date.today()
        self.now = datetime.now()
        self.current_date = str(
            self.now.strftime("%d/%m/%Y %H:%M:%S")).replace(' ',
                                                            '_').replace(
            '/', '_').replace(':', '_')
        self.header_name = f'{self.atp_name}_{self.func_name}_Date_{self.current_date}'
        self.target_folder = 'target_folder'

        if os.path.isfile(self.header_name) == False:
            file_name = f'{self.target_folder}{self.header_name}.txt'
            try:
                with open(file_name, 'w', encoding='utf-8') as f:
                    f.write(f'''ATP: {self.atp_name}\nFunction name: {self.func_name}\nError: {error_name}\nDate: {self.current_date}\n\n{trace_back}''')
            except Exception as e:
                print(e)

    def success_signal_connection(self):
        self.stop_flag = False
        print("\nFunction success")

另外,我发现当Worker调用包含Tk askopenfilename()的函数时总会出现问题,以下是报错回溯示例:

line 57, in run

**x = func()**

line 64, in open_main_terminal

**self.path1, self.file_name = upload_item()**

line 62, in upload_item

**filename = askopenfilename()**

line 375, in askopenfilename

**return Open(**options).show()**

line 39, in show

**w = Frame(self.master)**

line 2741, in __init__

**Widget.__init__(self, master, 'frame', cnf, {}, extra)**

line 2296, in __init__ (widgetName, self._w) + extra + self._options(cnf))

**RuntimeError: main thread is not in main loop**

内容的提问来源于stack exchange,提问作者Zohar Hazoom

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.31 15:21:02