Django中用Python multiprocessing从XML批量建模型连接关闭报错如何解决

• 数据库连接继承冲突：Django的数据库连接是进程级别的，主进程初始化的数据库连接会被fork出的子进程继承，子进程使用该继承连接时会和主进程的连接状态冲突，触发connection already closed报错
• 现有多进程逻辑不符合业务要求：你要求必须按顺序处理xml元素（元素间存在关联），但imap_unordered会乱序返回结果，完全破坏了关联处理逻辑，且你没有给create_profile传递所需的parent参数
• 性能瓶颈判断错误：你当前的性能问题大概率是数据库IO瓶颈（单条调用.create()插入数据，每次都要开启事务、写入、提交，开销极大），而非CPU瓶颈，盲目使用多进程不会解决核心问题
• 原有单进程代码存在变量名错误：参数定义为xml_members，弹出元素时用的是members.pop(0)，会触发未定义变量报错

方案1：优先使用批量插入（改造成本最低收益最高） #

把单条.create()改成先构建所有模型实例，再调用bulk_create批量插入，数万条数据的插入耗时可以从几十分钟降到几秒，无需引入多进程：

def create_profiles_from_xml(xml_members, device):
    profiles = []
    # 只构建Profile实例，不要调用.create()或.save()
    while len(xml_members) > 0:
        parent_member = xml_members.pop(0)
        profile = create_profile(parent_member, parent=None)
        if profile:
            profiles.append(profile)
    # 批量插入，batch_size可根据数据库配置调整为500~2000
    Profile.objects.bulk_create(profiles, batch_size=1000)
    return profiles

注意需要同步修改create_profile函数，去掉内部的.create()调用，改为直接返回未保存的Profile实例即可。

方案2：确需多进程时的正确实现 #

如果create_profile的字段匹配逻辑是CPU密集型，才需要搭配多进程处理，实现要满足3个要求：

1. 子进程初始化时关闭继承的旧数据库连接
2. 子进程仅处理CPU密集的实例构建逻辑，不操作数据库
3. 按关联规则拆分任务块，块间无关联可并行，块内按顺序处理

from multiprocessing import Pool, cpu_count
from django.db import connection

# 子进程初始化钩子，关闭继承的旧连接
def init_worker():
    connection.close()

# 子进程任务：纯CPU计算，构建实例不操作数据库
def process_block(args):
    member_block, parent = args
    block_profiles = []
    for member in member_block:
        profile = create_profile(member, parent=parent)
        if profile:
            block_profiles.append(profile)
    return block_profiles

def create_profiles_from_xml(xml_members, device):
    # 按你的关联规则拆分任务块，确保块之间无关联，块内顺序处理
    blocks = []
    current_block = []
    for member in xml_members:
        current_block.append(member)
        # 每1000条无关联的元素拆为一个块，拆分规则可根据你的业务调整
        if len(current_block) >= 1000:
            blocks.append((current_block, None))
            current_block = []
    if current_block:
        blocks.append((current_block, None))
    
    # 启动进程池处理
    all_profiles = []
    with Pool(processes=cpu_count(), initializer=init_worker) as pool:
        for block_result in pool.map(process_block, blocks):
            all_profiles.extend(block_result)
    
    # 主进程统一批量插入
    Profile.objects.bulk_create(all_profiles, batch_size=1000)
    return all_profiles

• 除非确认瓶颈在CPU逻辑，否则不要用多进程，进程间通信、连接管理会引入额外开销
• 如果模型有自增主键，Django 1.10+版本的bulk_create默认会返回带主键的实例，不影响后续关联操作

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