多线程创建含MySQL连接池的Thread子类对象时阻塞问题求助

我定义了一个继承自Thread的MyObject类，其__init__方法中会创建MySQL连接池。尝试异步创建该类对象时始终出现阻塞问题：
最初用列表推导式创建对象：

self.objects = [MyObject(*o, self) for o in objectData]

但随着实例数量增加，延迟不断累积。
于是改用多线程创建：

self.objects = []
threads = [
    threading.Thread(
        target=lambda o: self.objects.append(MyObject(*o, self)), 
        args=(o,)
    ) for o in objectData
]

for t in threads:
    t.start()
for t in threads:
    t.join()

尽管对象实例能立即创建，但MyObject中连接池的创建耗时仍随数量增加而变长。请问mysql-connector中是否存在导致并发执行阻塞的机制？该如何解决？

附MyObject代码：

class MyObject(Thread):
    def __init__(self, *args, manager):
        self.parent = manager

        self.cnxPool = mysql.connector.pooling.MySQLConnectionPool(
            pool_name=f"{get_ident()}",
            pool_size=10,
            host=HOST,
            user=USER,
            password=PASSWORD,
            port=25060
        )

        self.stopEvent = Event()
        Thread.__init__(self)
        self.daemon = True

    def run(self):
        ...

objectData从JSON加载，包含可变数量的参数。

解答 #

一、mysql-connector并发创建连接池阻塞的原因 #

• 全局锁竞争：mysql-connector内部维护了_CONNECTION_POOLS全局字典用于管理所有连接池，创建新池时会对该字典加全局互斥锁。并发场景下所有线程都会竞争这把锁，导致连接池创建串行化，耗时随实例数线性增长。
• 连接预初始化串行：每个连接池创建时默认会预建立pool_size个数据库连接，这是耗时的网络IO操作。mysql-connector的连接建立逻辑存在串行点，即使GIL在IO等待时释放，大量并发创建连接仍会累积延迟。
• 池命名冲突风险：用get_ident()作为池名，线程复用场景下可能出现ID重复，导致连接池创建失败或覆盖，额外增加锁等待与重试的耗时。

二、解决方法 #

1. 复用全局连接池
   放弃每个MyObject实例单独创建连接池的方案，改为全局共享一个或少量连接池，所有实例直接从全局池获取连接：

   # 全局初始化连接池，按需调整pool_size
   global_pool = mysql.connector.pooling.MySQLConnectionPool(
       pool_name="global_pool",
       pool_size=len(objectData)*10,
       host=HOST,
       user=USER,
       password=PASSWORD,
       port=25060
   )
   
   class MyObject(Thread):
       def __init__(self, *args, manager):
           self.parent = manager
           self.cnxPool = global_pool  # 复用全局池
           self.stopEvent = Event()
           Thread.__init__(self)
           self.daemon = True
   

   此方式彻底避免并发创建连接池的锁竞争，连接池仅初始化一次。

2. 延迟连接池初始化
   将连接池创建逻辑从__init__移到run()方法中，让连接池在线程启动后再初始化，分散初始化压力：

   class MyObject(Thread):
       def __init__(self, *args, manager):
           self.parent = manager
           self.cnxPool = None
           self.stopEvent = Event()
           Thread.__init__(self)
           self.daemon = True
   
       def run(self):
           # 线程启动后再创建连接池
           self.cnxPool = mysql.connector.pooling.MySQLConnectionPool(
               pool_name=f"pool_{self.ident}",
               pool_size=10,
               host=HOST,
               user=USER,
               password=PASSWORD,
               port=25060
           )
           # 后续业务逻辑
           ...
   

3. 关闭连接预初始化
   创建连接池时关闭预检测与预初始化，将连接建立延迟到实际业务使用时，减少初始化阶段的开销：

   self.cnxPool = mysql.connector.pooling.MySQLConnectionPool(
       pool_name=f"pool_{self.ident}",
       pool_size=10,
       host=HOST,
       user=USER,
       password=PASSWORD,
       port=25060,
       pool_reset_session=False,
       pool_pre_ping=False  # 关闭连接预检测
   )
   

4. 控制并发线程数
   使用线程池限制并发创建实例的数量，避免系统资源过度竞争：

   from concurrent.futures import ThreadPoolExecutor
   
   self.objects = []
   # 根据数据库承载能力设置max_workers
   with ThreadPoolExecutor(max_workers=8) as executor:
       for o in objectData:
           obj = executor.submit(MyObject, *o, manager=self).result()
           self.objects.append(obj)
   

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