基于pyodbc的多Pervasive数据库多线程SQL查询优化方案咨询

问题背景 #

当前脚本通过单线程逐个连接多个Pervasive数据库执行SQL查询，因IO阻塞导致总耗时过长，需要通过多线程实现并行查询，同时解决连接未及时关闭的问题。

核心优化思路 #

• pyodbc的连接/游标不是线程安全的，必须为每个线程分配独立的连接和游标
• 使用concurrent.futures.ThreadPoolExecutor管理线程池，自动处理线程的创建、调度和回收
• 每个线程独立完成「连接数据库→执行查询→处理结果→关闭连接」的完整流程，避免资源泄漏

完整实现代码 #

import pyodbc
import pandas as pd
import logging
from concurrent.futures import ThreadPoolExecutor, as_completed

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 配置参数
server = '1.1.1.1:111'
username = 'test'
password = 'test123'
list_of_databases = ['DB1', 'DB2', 'DB3']  # 替换为你的实际数据库列表
sql_query = "SELECT * FROM Table where Date between '20230226' and '20230227'"

def query_single_database(db_name):
    """单个数据库的查询逻辑，独立运行在单个线程中"""
    connection = None
    cursor = None
    try:
        # 构建完整连接字符串（包含账号密码）
        connect_string = (
            'DRIVER=Pervasive ODBC Interface;'
            'SERVERNAME={server};'
            'DBQ={db};'
            'UID={user};'
            'PWD={pwd}'
        ).format(server=server, db=db_name, user=username, pwd=password)
        
        # 建立连接和游标
        connection = pyodbc.connect(connect_string)
        cursor = connection.cursor()
        logger.info(f"成功连接数据库: {db_name}")
        
        # 执行查询并转换为DataFrame
        cursor.execute(sql_query)
        rows = cursor.fetchall()
        columns = [col[0] for col in cursor.description]
        df = pd.DataFrame.from_records(rows, columns=columns)
        logger.info(f"完成数据库 {db_name} 的查询，共获取 {len(df)} 条数据")
        return (db_name, df)
    
    except Exception as e:
        logger.error(f"数据库 {db_name} 处理失败: {str(e)}")
        return (db_name, None)
    
    finally:
        # 强制关闭资源，避免泄漏
        if cursor:
            cursor.close()
        if connection:
            connection.close()
        logger.info(f"已关闭数据库 {db_name} 的连接")

def main():
    all_results = []
    # 线程池大小建议：根据数据库数量和服务器性能调整，一般不超过CPU核心数*2或数据库总数
    with ThreadPoolExecutor(max_workers=5) as executor:
        # 提交所有数据库查询任务
        futures = {executor.submit(query_single_database, db): db for db in list_of_databases}
        
        # 实时处理完成的任务
        for future in as_completed(futures):
            db_name, result_df = future.result()
            if result_df is not None:
                all_results.append(result_df)
    
    logger.info(f"所有查询完成，共获取 {len(all_results)} 个数据库的有效数据")
    # 可在此处合并结果或做后续处理
    # combined_df = pd.concat(all_results, ignore_index=True)

if __name__ == "__main__":
    main()

关键细节说明 #

• 线程安全保障：每个线程独立创建连接和游标，彻底避免多线程共享连接引发的异常
• 资源强制回收：在finally块中关闭游标和连接，确保即使查询出错也不会泄漏数据库连接资源
• 线程池规模控制：max_workers不要设置过大，避免对数据库服务器造成过载压力，建议根据实际场景调整为5-10之间
• 容错性提升：每个线程单独捕获异常，单个数据库查询失败不会导致整个程序崩溃
• 结果实时处理：通过as_completed可以在任务完成时立即处理结果，无需等待所有任务结束

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