如何在Django测试框架中模拟并行请求以复现数据库写入竞态问题

你遇到的是典型的检查后执行（TOCTOU）竞态条件，和缓存无关：两个请求几乎同时到达时，会先后执行get查询，此时双方都没查到对应md5的记录，随后都会执行create操作，最终生成重复数据。哪怕你的视图加了atomic声明，只要没有数据库层面的唯一约束，PostgreSQL默认的读已提交隔离级别也无法阻止这类问题。

Django原生测试框架没有内置直接发起非阻塞并行请求的能力，你可以借助Python标准库的多线程能力实现，不需要引入额外依赖，注意不要用多进程实现：Django测试环境下多进程会连接到独立的数据库副本，无法复现同库的竞态条件。

具体测试实现步骤 #

1. 使用TransactionTestCase而非常规TestCase：常规TestCase会用单事务包裹整个测试逻辑，不同线程的数据库操作互相不可见，无法触发竞态。
2. 用concurrent.futures.ThreadPoolExecutor提交并行请求
3. 所有请求执行完成后统计重复条目数，验证竞态是否触发

测试代码示例 #

from django.test import TransactionTestCase, Client
from concurrent.futures import ThreadPoolExecutor
from your_app.models import DataFile

class DataFileRaceTest(TransactionTestCase):
    reset_sequences = True

    def test_duplicate_create_race(self):
        client = Client()
        test_params = {
            "md5": "test_duplicate_md5_abc123",
            "filename": "test_file.txt"
        }

        # 同时发起5个并行请求，提高竞态触发概率
        with ThreadPoolExecutor(max_workers=5) as pool:
            futures = [
                pool.submit(
                    client.post,
                    "/your/view/path/",
                    data=test_params
                )
                for _ in range(5)
            ]
            # 等待所有请求处理完成
            for future in futures:
                future.result()
        
        # 校验重复条目
        record_count = DataFile.objects.filter(md5=test_params["md5"]).count()
        # 竞态触发时record_count会大于1
        self.assertGreater(record_count, 1, "本次测试未触发竞态，可重试多次验证")

测试注意事项 #

• 竞态属于概率性问题，单次测试可能无法复现，可重复运行测试或调大线程数提高触发概率
• 给md5字段添加唯一约束后再运行该测试，会有部分请求抛出IntegrityError，最终record_count会等于1，可验证唯一约束的修复效果
• 你现有逻辑可以直接替换为Django内置的get_or_create方法，该方法内部会优先尝试创建，捕获唯一约束冲突后再查询，比你当前的先查后写逻辑更安全，前提是已经添加数据库唯一约束。

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