如何在Django中正确使用multiprocessing模块规避常见问题？

问题1：Spawn模式下Django模型初始化错误 #

Spawn模式下子进程不会继承主进程的Django上下文，必须手动完成初始化：

• 在子进程的任务函数开头，先设置Django配置环境变量，再调用django.setup()
• 必须在django.setup()之后再导入模型，禁止在模块级别提前导入

示例代码：

import os
import django
from multiprocessing import Process

def specific_task(task_params):
    # 初始化Django上下文
    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')
    django.setup()
    
    # 销毁主进程继承的无效数据库连接，让Django自动创建子进程专属连接
    from django.db import connection
    connection.close()
    
    # 安全导入模型并执行任务
    from your_app.models import TargetModel
    obj = TargetModel.objects.get(id=task_params['obj_id'])
    # 执行CPU密集型逻辑
    ...

if __name__ == '__main__':
    # websocket触发时，直接创建进程并传入参数
    task_args = {'obj_id': 123}
    p = Process(target=specific_task, args=(task_args,))
    p.start()

问题2：多进程数据库访问错误结果 #

核心问题是进程间共享了数据库连接或模型实例，解决思路如下：

1. 强制每个子进程使用独立连接：Django初始化后立即调用connection.close()，销毁主进程继承的连接，Django会自动为子进程创建新连接
2. 禁止进程间传递模型实例：只传递主键、ID等可序列化数据，子进程内部重新查询模型对象
3. 用事务和行锁保证原子性：保留默认的READ COMMITTED隔离级别，关键操作包裹在事务中，配合select_for_update()锁定目标行：

from django.db import transaction

def specific_task(task_params):
    # ... 初始化Django和连接 ...
    with transaction.atomic():
        obj = TargetModel.objects.select_for_update().get(id=task_params['obj_id'])
        # 执行修改操作
        obj.status = 'processing'
        obj.save()

Celery适配方案 #

如果想替代原生multiprocessing，Celery可以无缝适配你的场景：

1. 任务定义：每个特定任务对应一个Celery任务函数，内部同样需初始化Django（Celery默认会自动处理，手动初始化可确保兼容性）
2. 触发方式：websocket收到用户请求时，直接调用task.delay(task_params)，Celery会自动调度进程执行
3. 数据传输：仅传递可序列化参数（如ID、字典），禁止传递模型实例
4. 实时结果反馈：结合Redis Pub/Sub，任务完成后发布结果到指定频道，主进程的websocket监听频道并推送给用户

示例Celery任务：

from celery import Celery
import os
import django

app = Celery('your_project')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

@app.task
def celery_specific_task(task_params):
    # 初始化Django上下文
    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')
    django.setup()
    
    from your_app.models import TargetModel
    from django.db import transaction
    with transaction.atomic():
        obj = TargetModel.objects.select_for_update().get(id=task_params['obj_id'])
        # 执行CPU密集型逻辑
        ...
        # 任务完成后发布结果到Redis
        import redis
        r = redis.Redis()
        r.publish(f"task_result:{task_params['user_id']}", "任务完成")

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