PyQt5集成MySQL连接器时程序意外退出问题排查

在用Python开发数据仓库时，使用MySQL Connector作为数据库驱动，单独运行数据仓库代码无异常，但集成到PyQt5的UI界面后，调用数据仓库功能时程序会突然退出，以下是相关信息及问题分析：

数据仓库代码（db.py） #

import mysql.connector
from mysql.connector import Error
from mysql.connector import errorcode

class Database:
    def __init__(self):
        self.dbc = ("localhost","db", "user", "pass")

    def __enter__(self):
        try:
            self._conn = mysql.connector.connect(host = self.dbc[0], database = self.dbc[1], user = self.dbc[2] , password = self.dbc[3])
            self._cursor = self._conn.cursor(dictionary=True)
        except mysql.connector.Error as err:
            if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
                    print("Something is wrong with your user name or password")
            elif err.errno == errorcode.ER_BAD_DB_ERROR:
                    print("Database does not exist")
            else:
                    print(err)
            raise
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        try:
           self._cursor.close()
           self._conn.close()
        except AttributeError: # isn't closable
            print('Not closable.')
            return True # exception handled successfully

    def commit(self):
        self._conn.commit()

    def close(self, commit=True):
        if commit:
            self.commit()
        self._conn.close()

    def ping_connection(self):
        if not self._conn.is_connected():
            self._conn.reconnect()

    def fetch_all(self, query: str, params: tuple = None) :
        self.ping_connection()
        self._cursor.execute(query, params or ())
        result = self._cursor.fetchall() 
        return result  
        
    def get_list(self) :
            sql = """SELECT * from Users;"""
            return self.fetch_all(sql)
    
if __name__ == "__main__":
    with Database() as db:
            print(db.get_list())

错误信息 #

Exception ignored on threading shutdown:
Traceback (most recent call last):
  File "C:\Users\amuser\AppData\Local\Programs\Python\Python313\Lib\threading.py", line 1524, in _shutdown
    if _main_thread._handle.is_done() and _is_main_interpreter():
SystemError: <method 'is_done' of '_thread._ThreadHandle' objects> returned a result with an exception set

测试用QT界面代码 #

from PyQt5.QtWidgets import QApplication, QDialog, QPushButton, QMessageBox
import sys
from db import Database

class TestDialog(QDialog):
    def __init__(self):
        super(TestDialog, self).__init__()
        self.setWindowTitle("Test Dialog")

        # Create a button and set its position and size
        self.test_button = QPushButton("Click Me", self)
        self.test_button.setGeometry(50, 50, 100, 30)
        
        # Connect the button click event to the `on_button_click` method
        self.test_button.clicked.connect(self.on_button_click)

    def on_button_click(self):
        # Show a message box when the button is clicked
        with Database() as db:
            result = db.get_list()

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

已安装包列表 #

可能的原因及解决方案 #

• 主线程阻塞与线程冲突：PyQt的UI操作必须在主线程执行，直接在UI点击事件中执行数据库操作会阻塞主线程，同时MySQL Connector的自动重连等操作会启动后台线程，程序退出时线程清理易出现异常。
• 版本兼容性问题：使用的mysql-connector-python 9.1.0版本较新，可能与Python 3.13、PyQt5存在未适配的兼容性问题。
• 资源清理逻辑缺陷：Database类的__exit__方法中，资源清理的判断逻辑不足，可能因线程状态异常导致清理失败，触发线程shutdown错误。

具体修复步骤：

1. 将数据库操作移到子线程
   隔离数据库操作与UI主线程，避免阻塞和线程冲突：

   from PyQt5.QtCore import QThread, pyqtSignal
   from db import Database
   
   class DBWorker(QThread):
       finished = pyqtSignal(list)
       error = pyqtSignal(str)
   
       def run(self):
           try:
               with Database() as db:
                   result = db.get_list()
                   self.finished.emit(result)
           except Exception as e:
               self.error.emit(str(e))
   
   # 修改TestDialog中的方法
   def on_button_click(self):
       self.worker = DBWorker()
       self.worker.finished.connect(self.on_db_result)
       self.worker.error.connect(self.on_db_error)
       self.worker.start()
   
   def on_db_result(self, result):
       QMessageBox.information(self, "结果", f"查询到{len(result)}条数据")
   
   def on_db_error(self, err_msg):
       QMessageBox.critical(self, "错误", err_msg)
   

2. 降级MySQL Connector版本
   切换到更稳定的版本，避免新特性的兼容性问题：

   pip install mysql-connector-python==8.0.36
   

3. 优化资源清理逻辑
   完善__exit__方法的资源判断，确保连接和游标正确关闭：

   def __exit__(self, exc_type, exc_val, exc_tb):
       try:
           if hasattr(self, '_cursor') and self._cursor:
               self._cursor.close()
           if hasattr(self, '_conn') and self._conn.is_connected():
               self._conn.close()
       except Exception as e:
           print(f"资源清理失败: {e}")
       return exc_type is None
   

4. 调整连接重连逻辑
   移除自动ping重连，改为在查询失败时手动重建连接，减少后台线程创建：

   def fetch_all(self, query: str, params: tuple = None) :
       try:
           self._cursor.execute(query, params or ())
           return self._cursor.fetchall()
       except mysql.connector.Error as err:
           if err.errno in (errorcode.CR_SERVER_LOST, errorcode.CR_SERVER_GONE_ERROR):
               self._conn.reconnect()
               self._cursor = self._conn.cursor(dictionary=True)
               self._cursor.execute(query, params or ())
               return self._cursor.fetchall()
           else:
               raise
   

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