如何在Django TestCase中使用多数据库连接以复现死锁场景

你遇到的问题核心在于子进程继承了父进程的数据库连接，导致两个进程实际上共用同一个数据库会话。在同一个数据库会话里，select_for_update 不会对自身加互斥锁，自然没法复现跨进程的死锁场景。下面我给你具体的解决思路和修改后的代码：

问题原因分析 #

Django在测试用例初始化时会建立数据库连接，当你用multiprocessing.Process创建子进程时，子进程会直接继承父进程的这个连接。但数据库连接是进程隔离的，跨进程共享连接不仅不安全，还会导致所有操作都在同一个会话中执行——同一个会话内的select_for_update不会互相阻塞，所以你预期的死锁根本不会发生。

解决方案：让每个子进程独立建立数据库连接 #

要解决这个问题，只需要在每个子进程的处理函数开头，主动关闭继承的旧连接，Django会在需要执行数据库操作时自动创建新的独立连接。具体步骤如下：

• 在process_1和process_2函数的最开始，调用django.db.connection.close()关闭继承的连接
• 保留原有的事务装饰器，确保每个进程的操作在独立事务中执行

修改后的代码示例：

import time
from multiprocessing import Process
from django.db import connection
from django.db.transaction import atomic
from django.test import TestCase
from your_app.models import Change  # 替换成你的模型实际路径

@atomic()
def process_1():
    # 关闭继承自父进程的旧连接，强制Django创建新的独立连接
    connection.close()
    change = Change.objects.select_for_update().get(id="1")
    time.sleep(5)
    change = Change.objects.select_for_update().get(id="2")

@atomic()
def process_2():
    # 同样关闭继承的连接，建立自己的数据库会话
    connection.close()
    change = Change.objects.select_for_update().get(id="2")
    time.sleep(5)
    change = Change.objects.select_for_update().get(id="1")

class DeadlockReproductionTest(TestCase):
    def test_deadlock_occurrence(self):
        # 提前创建测试所需的数据，注意：如果用默认TestCase，需确保数据能被子进程看到
        # 若子进程看不到数据，建议改用TransactionTestCase
        Change.objects.create(id="1")
        Change.objects.create(id="2")
        
        p1 = Process(target=process_1)
        p2 = Process(target=process_2)
        p1.start()
        p2.start()
        p1.join()
        p2.join()
        
        # 注意：死锁发生时，数据库会终止其中一个事务，对应进程exitcode会非0
        # 你需要根据实际情况调整断言逻辑，比如捕获数据库异常或验证死锁触发
        # self.assertEqual(p1.exitcode, 0)
        # self.assertEqual(p2.exitcode, 0)

额外注意事项 #

• 测试数据可见性：默认的TestCase会在事务中运行测试，子进程无法看到父进程未提交的事务数据。如果遇到子进程找不到测试数据的问题，建议改用TransactionTestCase（它会在测试前后提交/回滚事务）。
• 死锁断言调整：当死锁真正发生时，数据库会自动终止其中一个事务，对应的进程会抛出OperationalError类异常，进程退出码会是非0值，所以你原来的exitcode == 0断言需要根据实际场景修改。

备注：内容来源于stack exchange，提问作者scanf3