如何在Django中正确搭配使用多进程与事务块

错误根因 #

Django的数据库连接是进程绑定的，multiprocessing生成的子进程会复制主进程的内存状态，包括已经打开的数据库连接，但复制过来的连接在子进程中不可用，你写的worker_init关闭旧连接的逻辑是正确的，子进程后续操作数据库会自动新建连接，但核心问题是：每个子进程的数据库事务完全独立，主进程的transaction.atomic块无法控制子进程内的数据库操作，所以你把进程池包裹在主进程的事务块中是无效的，子进程的写入操作会自动提交，主进程抛异常也无法回滚，同时连接状态错乱还会触发connection already closed报错。

适配场景的落地方案 #

你提到的两个场景（视图批量操作全量回滚、管理命令支持dry run）都可以用「计算与写入分离」的思路实现，完全避开跨进程事务的问题。

核心逻辑 #

1. 子进程仅负责业务逻辑计算，不执行任何数据库写入操作，只将需要创建/修改的模型实例、更新字段字典等可序列化数据返回给主进程
2. 所有子进程执行完成无报错后，主进程在自身的transaction.atomic块中统一执行所有数据库写入操作，出现异常或需要dry run时直接回滚即可

这种方案也适配你无法使用bulk_create的场景：继承模型就算逐个在主进程中save，性能也远高于边计算边写入，业务逻辑的计算才是性能瓶颈。

修正后的管理命令代码示例 #

from multiprocessing import cpu_count, Pool
from django.db import transaction, connection
from django.core.management.base import BaseCommand
# 替换为你自己的继承模型
from .models import MyInheritedModel

# 子进程仅做计算，不碰DB写入
def some_func(item):
    # 这里写你的业务逻辑，生成需要新增/修改的模型实例，不要调用save()
    new_instance = MyInheritedModel(
        field1=item["xxx"],
        field2=item["yyy"]
    )
    # 如果是更新操作，返回(主键id, 待更新字段字典)即可
    return [new_instance]

def worker_init():
    # 关闭子进程继承的主进程无效连接，避免连接混乱
    connection.close()

class Command(BaseCommand):
    def add_arguments(self, parser):
        parser.add_argument(
            "--commit",
            action="store_true",
            default=False,
            help="是否实际提交修改到数据库"
        )

    def handle(self, *args, **options):
        commit = options["commit"]
        items = [...] # 替换为你要处理的任务列表

        try:
            # 第一步：多进程并行执行计算逻辑
            results = []
            with Pool(processes=cpu_count(), initializer=worker_init) as pool:
                for result in pool.imap_unordered(some_func, items):
                    results.extend(result)
            
            # 第二步：主进程统一在事务中执行所有DB写入
            with transaction.atomic():
                for instance in results:
                    # 继承模型不支持bulk_create就逐个save
                    instance.save()
                    # 如果是更新操作，对应写法：
                    # MyInheritedModel.objects.filter(id=item_id).update(**update_dict)
                
                # dry run场景直接触发回滚
                if not commit:
                    raise RuntimeError("DRY RUN_TRIGGER_ROLLBACK")
            
            self.stdout.write(self.style.SUCCESS("操作执行完成，数据已提交"))
        except RuntimeError as e:
            if "DRY RUN_TRIGGER_ROLLBACK" in str(e):
                self.stdout.write(self.style.WARNING("DRY RUN: 脚本运行成功，未修改任何数据"))
            else:
                raise
        except Exception as e:
            self.stdout.write(self.style.ERROR(f"操作失败，所有修改已回滚：{str(e)}"))

特殊场景适配 #

如果你的业务逻辑中必须在子进程中查询数据库，是可以正常操作的：worker_init关闭旧连接后，子进程查询时会自动新建独立连接，只要不在子进程中执行写入操作即可。

如果确实需要在子进程中执行写入，建议改成两阶段执行：

1. 所有子进程先跑一遍dry run，仅做逻辑校验，全部执行成功后再走正式写入流程
2. 确保子进程的写入逻辑是幂等的，避免部分进程执行失败导致数据不一致

视图场景的额外注意点 #

视图中不建议直接使用multiprocessing，很容易导致服务器进程资源耗尽，同时请求超时限制也无法支撑大量数据的计算写入。如果计算量较大，建议换成Celery异步任务处理，计算完成后再回调通知用户结果。

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