Python多进程执行DB语句：控制最大n并行调用及代码报错修复

原代码的核心问题 #

1. 切片越界：else分支里的statement_list[n:no_of_threads]是错误的，应该取n到n+no_of_threads的区间，否则当n大于no_of_threads时，切片范围无效，直接触发IndexError。
2. 进程数设置错误：循环里创建Pool时用了processes=n，初始n=0会导致创建0个进程，完全无法执行任务，应该固定使用no_of_threads作为进程数。
3. 函数传递错误：db.sqlscript()是直接调用方法，而非传递方法对象给map，map需要接收可调用函数，这里应该写db.sqlscript（不带括号）。
4. 跨进程连接共享风险：数据库连接对象不能跨进程共享，原代码直接传递db对象给多进程，会导致连接异常或数据错乱。
5. 资源释放不当：循环内重复创建Pool却未正确关闭，且提前调用db.close()会导致后续任务无法使用连接。

优化后的实现方案 #

考虑到多进程中数据库连接的安全性，我们让每个进程独立创建数据库连接，避免共享连接的问题。以下是修正后的代码：

import psycopg2
from multiprocessing import Pool

def execute_single_sql(sql_stmt, db_config):
    # 每个进程独立创建数据库连接
    conn = None
    result = None
    try:
        conn = psycopg2.connect(**db_config)
        cur = conn.cursor()
        cur.execute(sql_stmt)
        # 根据需求获取结果，有返回结果则fetchall，无返回则设为None
        result = cur.fetchall() if cur.description else None
        conn.commit()
    except Exception as e:
        if conn:
            conn.rollback()
        result = str(e)
    finally:
        if conn:
            conn.close()
    return result

def parallel_execute_db(db_config, statement_list, no_of_processes=10):
    # 根据任务数量和进程数确定实际使用的进程数
    actual_processes = min(no_of_processes, len(statement_list))
    if actual_processes <= 0:
        return []
    
    # 用with语句自动管理Pool生命周期，避免手动关闭的麻烦
    with Pool(processes=actual_processes) as pool:
        # 使用starmap传递多参数，把每个SQL和数据库配置配对
        results = pool.starmap(execute_single_sql, [(stmt, db_config) for stmt in statement_list])
    
    return results

# 使用示例
if __name__ == "__main__":
    db_config = {
        "host": "your_host",
        "user": "your_user",
        "password": "your_password",
        "dbname": "your_db"
    }
    sql_list = ["SELECT * FROM table1", "INSERT INTO table2 VALUES (1, 'test')"]
    results = parallel_execute_db(db_config, sql_list, no_of_processes=10)
    print(results)

优化点说明 #

• 独立连接：每个进程在执行SQL时自己创建/关闭连接，彻底避免跨进程共享连接的问题。
• 高效Pool管理：使用with语句自动管理Pool的生命周期，无需手动关闭。
• 动态进程数：自动取任务数和设定进程数的较小值，避免创建多余进程。
• 错误处理：添加异常捕获与回滚逻辑，保证每个SQL执行的可靠性。
• 参数传递：通过starmap传递数据库配置和SQL语句，适配多参数需求。

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