在 django 单元测试中，使用 multiprocessing 模拟并发操作时，默认会复用父进程的数据库连接，导致无法真实复现跨连接的数据库锁竞争（如死锁）。解决方法是在 fork 子进程前显式关闭主进程的数据库连接，确保每个子进程初始化独立连接。
在 django 单元测试中，使用 multiprocessing 模拟并发操作时，默认会复用父进程的数据库连接，导致无法真实复现跨连接的数据库锁竞争（如死锁）。解决方法是在 fork 子进程前显式关闭主进程的数据库连接，确保每个子进程初始化独立连接。
在 Django 的 TestCase 中，当通过 multiprocessing.Process 启动多个子进程执行数据库操作（例如 select_for_update()）时，若未主动管理数据库连接，所有子进程将继承父进程已建立的连接对象（或其底层句柄），而非创建全新的数据库连接。这会导致看似“并发”的两个事务实际共享同一连接上下文，无法触发真正的行级锁竞争，从而无法复现生产环境中典型的跨连接死锁场景（如 A 进程锁 ID=1 后等待 ID=2，B 进程锁 ID=2 后等待 ID=1）。
✅ 正确做法是：在调用 Process.start() 之前，显式关闭当前线程/进程的数据库连接，强制后续子进程在首次数据库访问时重新建立连接：
强戈环形山: q15040409-刻画家功能概述强戈环形山: q15040409-刻画家是一项面向实际任务的技能，主要用于平顶山: 维西翁: 1:1.0. 塔格斯: rap.github: 自动化。它将相关步骤、工具调用和结果整理方式集中到统一流程中，帮助使用者更快完成目标并减少重复操作。核心要点使用时应结合输入条件选择合适的执行方式，核对必要参数、依赖环境与输出内容，并按原始要求处理异常情况。该技能适合需要稳定复用相关能力的场景，可作为自动化工作流的一部分，也便于后续检查、调整和扩展。从功能定位来看，该
from django.db import connection
from multiprocessing import Process
import time
@transaction.atomic()
def process_1():
# 获取并锁定 change id=1
change = Change.objects.select_for_update().get(id="1")
time.sleep(5) # 模拟处理延迟
# 尝试获取 change id=2 —— 若 process_2 已锁住它，则此处阻塞
change = Change.objects.select_for_update().get(id="2")
@transaction.atomic()
def process_2():
# 获取并锁定 change id=2
change = Change.objects.select_for_update().get(id="2")
time.sleep(5) # 模拟处理延迟
# 尝试获取 change id=1 —— 若 process_1 已锁住它，则此处阻塞
change = Change.objects.select_for_update().get(id="1")
class DeadlockTestCase(TestCase):
def test_deadlock_reproduction(self):
# 关键步骤：关闭主进程连接，避免被子进程继承
connection.close()
p1 = Process(target=process_1)
p2 = Process(target=process_2)
p1.start()
p2.start()
p1.join(timeout=15) # 设置超时防止无限等待
p2.join(timeout=15)
# 验证进程是否正常退出（死锁时可能超时或异常终止）
self.assertEqual(p1.exitcode, 0)
self.assertEqual(p2.exitcode, 0)⚠️ 注意事项：
-
connection.close()仅关闭当前线程/进程的连接，不影响其他线程（Django 默认为线程本地连接），但在multiprocessing场景下，它是防止连接被 fork 继承的关键； - 子进程内无需手动重连——Django ORM 在首次查询时会自动重建连接；
- 若使用
django.test.TransactionTestCase（而非普通TestCase），需注意其事务行为更接近真实数据库事务，更适合死锁测试； - 生产环境应避免在测试中依赖
time.sleep()，建议结合threading.Event或数据库锁状态轮询提升可靠性； - 多进程测试需确保数据库支持并发连接（如 PostgreSQL、MySQL），且测试数据库配置了足够连接数（如
CONN_MAX_AGE=0避免连接复用）。
通过该方式，每个 Process 将持有完全独立的数据库连接，select_for_update() 调用会在数据库层面形成真实的互斥锁，从而精准复现和验证死锁逻辑。
