Python如何对列表元素实现多进程并行独立处理MySQL更新

• 待处理列表共1000个独立值，单值的数据库更新操作互不依赖
• 现有串行实现逐值建立/销毁MySQL连接、执行更新，执行效率偏低
• 核心需求为通过多进程并行处理提升效率，同时确认joblib相关API的可行性

你贴的串行代码本身存在逻辑错误：SQL语句里的each是Python变量，直接写在字符串中不会被替换，执行时会直接把each当成Name字段的匹配值，完全达不到预期效果，后续改造会统一修正这个问题，改用参数化查询实现。

不需要安装第三方依赖，直接用Python自带的多进程模块即可实现，核心逻辑是把单条数据的处理逻辑抽成独立函数，每个子进程独立创建数据库连接、执行操作、关闭连接，避免跨进程传递连接对象的问题。
示例代码：

import mysql.connector
from multiprocessing import Pool

# 单条数据处理逻辑
def process_single_item(name_val):
    mydb = None
    cs = None
    try:
        # 每个子进程内部独立创建连接
        mydb = mysql.connector.connect(
            host='localhost',
            database='College',
            user='root',
        )
        cs = mydb.cursor()
        # 参数化查询，避免SQL注入和语法错误
        statement = "UPDATE STUDENT SET AGE = 23 WHERE Name=%s"
        cs.execute(statement, (name_val,))
        mydb.commit()
        return (name_val, "success")
    except Exception as e:
        if mydb:
            mydb.rollback()
        return (name_val, f"failed: {str(e)}")
    finally:
        # 确保资源释放
        if cs:
            cs.close()
        if mydb:
            mydb.close()

if __name__ == "__main__":
    # 替换成你的实际待处理列表
    value_list = ["name1", "name2", "name3"] # 共1000个值
    # 并行进程数，根据数据库负载调整，不要超过MySQL最大连接数限制
    worker_num = 8
    with Pool(processes=worker_num) as pool:
        results = pool.map(process_single_item, value_list)
    # 统计执行结果
    success_cnt = len([r for r in results if r[1] == "success"])
    print(f"处理完成，成功{success_cnt}条，失败{len(value_list)-success_cnt}条")

你提到的joblib方案完全可以实现需求，代码写法比标准库更简洁，并行逻辑的调度封装更友好，同样需要注意不要跨进程传递数据库连接。
示例代码：

import mysql.connector
from joblib import Parallel, delayed

# 单条处理逻辑和上述方案一致
def process_single_item(name_val):
    mydb = None
    cs = None
    try:
        mydb = mysql.connector.connect(
            host='localhost',
            database='College',
            user='root',
        )
        cs = mydb.cursor()
        statement = "UPDATE STUDENT SET AGE = 23 WHERE Name=%s"
        cs.execute(statement, (name_val,))
        mydb.commit()
        return (name_val, "success")
    except Exception as e:
        if mydb:
            mydb.rollback()
        return (name_val, f"failed: {str(e)}")
    finally:
        if cs:
            cs.close()
        if mydb:
            mydb.close()

if __name__ == "__main__":
    value_list = ["name1", "name2", "name3"] # 替换为实际1000个值的列表
    worker_num = 8
    # n_jobs指定并行进程数，backend="loky"为默认多进程后端
    results = Parallel(n_jobs=worker_num, backend="loky")(
        delayed(process_single_item)(item) for item in value_list
    )
    success_cnt = len([r for r in results if r[1] == "success"])
    print(f"处理完成，成功{success_cnt}条，失败{len(value_list)-success_cnt}条")

• 绝对不要跨进程传递数据库连接、cursor对象：这些对象不是进程安全的，跨进程复用会出现随机报错、数据错乱、连接泄漏问题，必须在每个子进程的任务函数内部独立创建、销毁连接。
• 并行度不要盲目设高：数据库更新属于IO密集型操作，不需要把进程数开到和CPU核数一致，建议先从4~8开始测试，同时确认MySQL的max_connections配置，避免并行开太多把数据库连接打满，影响其他业务。
• 一定要做异常捕获：单条数据处理失败（比如Name不存在、连接超时）不要让整个任务崩溃，捕获异常后记录失败条目，后续可以单独重试。
• 优先用参数化查询：不要把变量直接拼接进SQL语句，既避免SQL注入风险，也能处理名字里带引号、特殊字符的情况，避免SQL语法错误。

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