Python MySQL连接器循环执行SQL查询时出现Connection timed out错误如何解决

问题根因 #

你当前的数据库操作逻辑存在明显的资源浪费问题：每次执行SQL查询都会独立创建、销毁一次数据库连接，1900次循环累计产生近4000次短连接请求。短时间内大量TCP连接处于TIME_WAIT状态、MySQL服务端可用连接被占满，就会出现随机超时的故障。

修复方案 #

1. 优先做批量查询优化（性能提升最明显） #

不需要循环逐次查询每个用户的信息，你可以先拿到所有用户hash的列表，一次性查询所有符合条件的用户数据，仅需1次数据库请求就能拿到全部结果，从根源上避免频繁连接的问题：

# 示例代码：先获取所有用户hash列表
all_user_hashes = [row[0] for row in perform_db_query_fetchall('SELECT hash FROM users WHERE fake = "no"', ())]
# 构造IN查询语句
placeholders = ', '.join(['%s'] * len(all_user_hashes))
query = f'SELECT `hash`, `portfolio`, `date`, `type`, `invested` FROM `users` WHERE `hash` IN ({placeholders}) AND `fake` = "no"'
all_user_data = perform_db_query_fetchall(query, tuple(all_user_hashes))
# 把结果转为字典，方便循环中按hash取值
user_data_map = {}
for row in all_user_data:
    user_hash = row[0]
    if user_hash not in user_data_map:
        user_data_map[user_hash] = []
    user_data_map[user_hash].append(row[1:])

# 后续遍历用户hash的时候直接从user_data_map取数据即可，不需要再查数据库

2. 复用数据库连接，避免每次查询新建连接 #

如果确实需要逐次查询，把连接初始化逻辑移到循环外部，整个批量任务共用同一个数据库连接，仅在所有操作完成后关闭，同时增加连接重试机制，避免偶发故障直接中断程序：

# 修改后的helpers.py示例，支持传入已有的连接
import mysql.connector
from config import *
import time

def get_db_connection():
    # 增加3次连接重试逻辑
    retry_count = 3
    for i in range(retry_count):
        try:
            return mysql.connector.connect(
                host=DatabaseConfig.host_db,
                user=DatabaseConfig.user_db,
                password=DatabaseConfig.password_db,
                database=DatabaseConfig.database_db
            )
        except mysql.connector.errors.InterfaceError:
            if i == retry_count -1:
                raise
            time.sleep(1)

def perform_db_query_fetchall(query: str, params: tuple, conn=None):
    # 支持传入外部连接，没有传入就新建临时连接
    local_conn = conn if conn else get_db_connection()
    c = local_conn.cursor()
    c.execute(query, params)
    results = c.fetchall()
    c.close()
    # 如果是外部传入的连接，不关闭，由调用方统一关闭
    if not conn:
        local_conn.close()
    return results

# 调用示例：循环外初始化连接，循环内复用
db_conn = get_db_connection()
for user_hash in all_user_hashes:
    # 传入已有连接，不新建
    user_coins_from_db = perform_db_query_fetchall('SELECT `portfolio`, `date`, `type`, `invested` FROM `users` WHERE `hash` = %s AND `fake` = "no"', (user_hash, ), conn=db_conn)
# 所有循环结束后统一关闭连接
db_conn.close()

3. 线上服务场景使用连接池 #

如果你的代码是Flask线上服务，不是单次运行的脚本，可以使用mysql-connector自带的连接池功能，固定复用少量连接，避免频繁建连的开销：

# 初始化连接池，全局只需初始化一次
from mysql.connector import pooling
db_pool = pooling.MySQLConnectionPool(
    pool_name="mypool",
    pool_size=5,
    host=DatabaseConfig.host_db,
    user=DatabaseConfig.user_db,
    password=DatabaseConfig.password_db,
    database=DatabaseConfig.database_db
)

# 查询时从连接池取连接
def perform_db_query_fetchall(query: str, params: tuple):
    conn = db_pool.get_connection()
    c = conn.cursor()
    c.execute(query, params)
    results = c.fetchall()
    c.close()
    conn.close() # 这里的close是把连接归还到连接池，不是真正销毁
    return results

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