Python mysql.connector多线程连接问题及连接池报错求助

问题背景 #

在多线程环境中使用包含MySQL连接的对象时，遇到MySQL Connection not available错误，确认是多线程导致。测试代码如下：

threads = []

with MysqlExample({ "host" : "XXXX", 
                              "database" : "XXXX",
                              "user" : "XXXX", 
                              "password" : "XXXX" }) as my_example:
    for _ in range(10): 
        th = threading.Thread( target = my_example.example3,
                                   args = (), )
        th.start()
        threads.append( th ) 
        
    for th in threads : 
        th.join() 

对应的对象代码：

class MysqlExample(object):

    def __init__(self, config : dict ):
        self._conn = mysql.Connect( **config )
        self._conn.autocommit = True
    
    def close(self):
        self._example2()
        self._conn.close()
    
    def __enter__(self):
        self._example1()
        return self
        
    def __exit__(self, type, value, tb ):
        self.close()
    
    def _example1(self):
        cur = self._conn.cursor()
        cur.execute( '''select * from simple_test''' )
        cur.fetchall()
        
    def _example2(self):
        cur = self._conn.cursor()
        cur.execute('''select * from simple_test''' )
        cur.fetchall()
        
    def example3(self):
        cur = self._conn.cursor()
        cur.execute('''select * from simple_test''' )
        cur.fetchall()

尝试过在每个方法中使用Lock()但无效，询问是否可用连接池解决。

更新问题：连接池尝试后出现新错误 #

使用连接池后出现ReferenceError: weakly-referenced object no longer exists错误，代码如下：

class MysqlExamplePooling(object):

    def __init__(self, config : dict ):
        self._pool = MySQLConnectionPool(pool_name = "MysqlExamplePooling",
                                                      pool_size = 3,
                                                      **config)
    
    def _conn(self): 
        cnx = self._pool.get_connection()
        cnx.autocommit = True
        return cnx
    
    def close(self):
        self._example2()
        self._pool.close()
    
    def __enter__(self):
        self._example1()
        return self
        
    def __exit__(self, type, value, tb ):
        self.close()
    
    def _example1(self):
        cur = self._conn().cursor()
        cur.execute( '''select * from simple_test''' )
        cur.fetchall()
        
    def _example2(self):
        cur = self._conn().cursor()
        cur.execute('''select * from simple_test''' )
        cur.fetchall()
        
    def example3(self):
        cur = self._conn().cursor()
        cur.execute('''select * from simple_test''' )
        cur.fetchall()
        

错误原因分析 #

1. 初始代码错误原因：
   MySQL连接对象不是线程安全的，多个线程共用同一个_conn对象时，会出现连接状态冲突——比如一个线程在使用连接时，另一个线程可能触发连接关闭或状态变更，直接导致MySQL Connection not available错误。
   用Lock()无效的原因通常是锁的范围不对：要么只锁了cursor创建/执行步骤，但连接本身的状态在锁外被干扰；要么锁的实现存在漏洞，没完全覆盖连接的所有操作。

2. 连接池代码错误原因：

   • 获取连接后未正确归还到池中：_conn()返回的连接对象在方法执行完毕后被垃圾回收，但连接池仍持有它的弱引用，引发ReferenceError。
   • 提前关闭连接池：__exit__中调用_pool.close()会关闭池内所有连接，若此时还有线程在使用从池获取的连接，必然触发错误。

解决方法 #

方案1：不用连接池，给每个线程分配独立连接 #

确保每个线程拥有专属的MySQL连接，避免共用：

class MysqlExample(object):
    def __init__(self, config : dict ):
        self.config = config
        # 不在初始化时创建连接，由每个线程单独创建
    
    def _get_conn(self):
        conn = mysql.Connect(**self.config)
        conn.autocommit = True
        return conn
    
    def example3(self):
        conn = self._get_conn()
        try:
            cur = conn.cursor()
            cur.execute('''select * from simple_test''' )
            cur.fetchall()
        finally:
            conn.close() # 用完立即关闭连接

测试代码无需修改，每个线程调用example3时会创建自己的连接，用完即关闭，无状态冲突。

方案2：正确实现连接池 #

核心是确保连接用完后归还到池，且不在线程运行时关闭池：

from mysql.connector.pooling import MySQLConnectionPool

class MysqlExamplePooling(object):
    def __init__(self, config : dict ):
        self._pool = MySQLConnectionPool(
            pool_name = "MysqlExamplePooling",
            pool_size = 10, # 建议匹配线程数，避免线程等待获取连接
            **config
        )
    
    def close(self):
        # 不在上下文管理器中关闭池，待所有线程执行完毕后再调用
        self._pool.close()
    
    def __enter__(self):
        return self
        
    def __exit__(self, type, value, tb ):
        # 此处不关闭池，留到所有线程join后再处理
        pass
    
    def _execute_query(self):
        cnx = None
        try:
            cnx = self._pool.get_connection()
            cur = cnx.cursor()
            cur.execute('''select * from simple_test''' )
            cur.fetchall()
        finally:
            if cnx:
                cnx.close() # mysql.connector中调用close()是归还连接到池，不是销毁
    
    def example3(self):
        self._execute_query()

测试代码调整为在所有线程结束后关闭池：

threads = []

with MysqlExamplePooling({ "host" : "XXXX", 
                              "database" : "XXXX",
                              "user" : "XXXX", 
                              "password" : "XXXX" }) as my_example:
    for _ in range(10): 
        th = threading.Thread( target = my_example.example3, args = (), )
        th.start()
        threads.append( th ) 
        
    for th in threads : 
        th.join() 
        
my_example.close() # 所有线程执行完毕后关闭池

额外注意事项 #

• 连接池pool_size不要超过MySQL服务器允许的最大连接数（默认一般为151），同时建议匹配线程数量，减少线程等待时间。
• 所有数据库操作必须放在try...finally块中，确保连接无论成功失败都能归还给池。
• 若使用上下文管理器，必须保证所有线程执行完毕后再关闭连接池，避免提前销毁连接。

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