Django 并发实战：用数据库事务与行锁防止竞态条件

假设一个账号只剩最后一次生成额度，你在两个浏览器标签页里几乎同时提交了请求，应用把两个都收下了。每个请求单独看都通过了余额检查，可应用接受的工作量已经超过了余额能支付的范围。这就是竞态条件（race condition）。本文用一个 Django 积分系统把这个问题拆开：先复现 bug，再用数据库事务与行级锁修复，最后测试两个请求争抢最后一分额度时的行为。适合已经会写 Django 模型、迁移和视图、但还没接触过并发处理的开发者。

准备工作

先确认环境与依赖。示例使用 Python 3.12、Django 5.2、Django REST Framework 3.16 和 PostgreSQL 17。数据库之所以选 PostgreSQL，是因为 SQLite 没有实现下面要用到的行锁。Docker 与 Compose 用来跑本地数据库。版本号可能随时间变化，安装前请以各项目官网当前信息为准。

整个教程里图像生成都是模拟的：接口接收一个 prompt，返回一个模拟结果，不需要任何 AI 服务商账号或付费密钥，练习的重点始终是额度判断和它引发的数据库变更。

先理解两个概念

并发指两个或多个任务在时间上有重叠地推进。服务器可以在请求 A 等待数据库响应时开始处理请求 B，两者的指令不必在同一瞬间执行。事件循环可以在某个任务等待时切换到另一个任务，这是 Python 支持并发的一种方式。关键在于：第一个操作还没结束，另一个操作就已经在推进了。

竞态条件是一种缺陷，结果的正确性取决于并发操作的时序或顺序。当多个操作共享数据、而应用没有充分协调它们的访问时就会出现。每个操作单独看都正确，但其中一个可能基于另一个已经改过的信息行动，换一个执行顺序就会得到不同的、错误的结果。并发提供了重叠的机会，竞态条件则是应用处理这种重叠时可能产生的 bug。

竞态条件还会出现在哪里

• 浏览量计数器丢更新。两个请求都读到 100，各自加一后都写回 101。计数本应是 102，一次更新凭空消失。这跟库存、购买都无关。
• 两个注册抢同一个用户名。如果只依赖一次前置的用户名可用性检查，两个请求可能都查到「可用」，然后各自创建账号。防护手段是在数据库层面强制唯一性，而不是相信那次检查。
• 两个 worker 领同一个任务。两个后台 worker 都在对方认领之前读到任务状态，都认为任务空闲并开始处理，结果通知发了两遍、报表生成了两份。认领机制必须在开工前协调归属。
• 旧的搜索响应覆盖新的。你输入「Django」，又改成「Django transactions」，第二个响应先回来并显示了正确结果，第一个响应随后到达并覆盖了页面。这种情况要用请求标识或过期响应检查来处理，数据库行锁不是对症的解法。

额度检查为什么会失败

先定下规则：每个图像请求消耗一分额度。额度代表应用的使用配额，与模型处理的 token 是两回事。这是本文为示例设定的规则。

如果账号初始有 10 分，接受 9 个请求后还剩 1 分。要接受一个请求，后端必须依次完成：读取账号余额、判断是否至少还剩 1 分、扣掉 1 分并保存、记录这次被接受的请求。

一次只测一个请求时，这个流程看起来完全正确：第一个请求把余额存成 0，下一个读到 0 就停下。接下来看同样的操作在时间上重叠时会发生什么。

假设请求 A 读到账号还剩 1 分。在它更新数据库之前，请求 B 读到了同一个账号，也看到 1 分。此时两个请求各自持有一份余额副本，都通过了检查，都算出 1 - 1 = 0。请求 A 保存 0 并记录一次生成，请求 B 随后也保存 0 并再记录一次生成。最终余额是 0，但应用接受了两个请求。检查余额是否为负并不能发现这类故障——真正的错误在于，在另一个请求有机会修改数据之后，仍然信任那个读到的值。

操作步骤

第一步：创建项目与虚拟环境

建目录、建虚拟环境并激活。下面的激活命令适用于 Linux 和 macOS：

mkdir django-credit-demo
cd django-credit-demo
python3.12 -m venv .venv
source .venv/bin/activate

在 Windows 上用 py -3.12 -m venv .venv 创建环境，用 .venv\Scripts\Activate.ps1 在 PowerShell 中激活。后续命令都要在这个环境处于激活状态时执行。

创建 requirements.txt：

Django>=5.2,<5.3
djangorestframework>=3.16,<3.17
psycopg[binary]>=3.2,<3.3

Django 提供模型和数据库工具，Django REST Framework 负责接口，Psycopg 提供 PostgreSQL 连接，[binary] 表示使用预编译实现。每个版本区间允许在当前发布系列内升级，但不跨到下一个系列。

安装依赖并创建项目与应用：

python -m pip install -r requirements.txt
python -m django startproject config .
python manage.py startapp credits

startproject 创建 config 包和 manage.py，末尾的点表示在当前目录生成；startapp 创建 credits 包，模型、服务函数、视图和测试都放在这里。

第二步：启动 PostgreSQL

在 manage.py 旁边创建 compose.yaml：

services:
  db:
    image: postgres:17
    environment:
      POSTGRES_DB: credit_demo
      POSTGRES_USER: credit_demo
      POSTGRES_PASSWORD: local-demo-only
    ports:
      - "127.0.0.1:5433:5432"
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U credit_demo -d credit_demo"]
      interval: 2s
      timeout: 5s
      retries: 15

端口映射把数据库只暴露在本机回环地址的 5433 端口上。健康检查每 2 秒执行一次 pg_isready，单次超时 5 秒，最多重试 15 次后判定容器不健康。启动命令：

docker compose up -d --wait

-d 让服务在后台运行，--wait 会等到健康检查通过才返回。整个教程期间保持这个容器运行。

第三步：配置 Django

用下面的最小配置替换 config/settings.py：

import os

SECRET_KEY = "local-tutorial-only-do-not-use-in-production"
DEBUG = True
ALLOWED_HOSTS = ["localhost", "127.0.0.1", "testserver"]

INSTALLED_APPS = [
    "django.contrib.auth",
    "django.contrib.contenttypes",
    "rest_framework",
    "credits",
]

MIDDLEWARE = []
ROOT_URLCONF = "config.urls"
DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField"
USE_TZ = True

DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.postgresql",
        "NAME": os.environ.get("DB_NAME", "credit_demo"),
        "USER": os.environ.get("DB_USER", "credit_demo"),
        "PASSWORD": os.environ.get("DB_PASSWORD", "local-demo-only"),
        "HOST": os.environ.get("DB_HOST", "127.0.0.1"),
        "PORT": os.environ.get("DB_PORT", "5433"),
        "OPTIONS": {"options": "-c lock_timeout=5000 -c statement_timeout=10000"},
    }
}

REST_FRAMEWORK = {
    "DEFAULT_AUTHENTICATION_CLASSES": [
        "rest_framework.authentication.BasicAuthentication",
    ],
    "DEFAULT_PERMISSION_CLASSES": [
        "rest_framework.permissions.IsAuthenticated",
    ],
}

几个要点：INSTALLED_APPS 启用 Django 的用户支持、DRF 和 credits 应用；空的 MIDDLEWARE 让这个 API 示例保持最小；ALLOWED_HOSTS 接受本地地址和测试客户端的主机名。数据库配置里每个 os.environ.get() 读取可选环境变量并回退到容器中的对应值。OPTIONS 设置了 5 秒锁超时和 10 秒语句超时，让卡住的操作直接报错而不是无限等待——超时是一条错误路径，这个精简接口没有为它提供自定义响应。

DRF 配置要求请求必须经过认证：BasicAuthentication 读取请求中携带的凭据，IsAuthenticated 在视图接受 prompt 之前就拒绝匿名调用者。示例中的密钥和 DEBUG = True 只适合本地开发，部署时要换成生产密钥、合适的认证方式和加密连接。

暂时把 config/urls.py 写成空路由列表，后面再补上接口路由。

第四步：定义模型

在 credits/models.py 中定义账号余额与生成记录。账号与 Django 用户一对一关联，余额用整数表示；每次被接受的请求写一条生成记录，便于事后核对到底接受了几次。

from django.conf import settings
from django.db import models


class CreditAccount(models.Model):
    user = models.OneToOneField(
        settings.AUTH_USER_MODEL,
        related_name="credit_account",
    )
    balance = models.IntegerField(default=0)


class Generation(models.Model):
    account = models.ForeignKey(
        CreditAccount,
        related_name="generations",
    )
    prompt = models.TextField()
    created_at = models.DateTimeField(auto_now_add=True)

生成迁移并应用：

python manage.py makemigrations credits
python manage.py migrate

第五步：写出有 bug 的版本

先故意写出会出错的实现，这样后面才能观察到问题。在 credits/services.py 中写一个「先读、再判断、再写」的函数：

from .models import CreditAccount, Generation


def request_generation_buggy(account_id, prompt):
    account = CreditAccount.objects.get(id=account_id)
    if account.balance < 1:
        return None
    account.balance -= 1
    account.save(update_fields=["balance"])
    return Generation.objects.create(account=account, prompt=prompt)

这段代码在单请求场景下完全正确，问题出在读取和保存之间存在时间窗口：另一个请求可以在这个窗口里读到同一个旧余额。

第六步：复现竞态条件

写一个测试，让两个线程同时调用上面这个函数，账号初始只有 1 分。测试用 TransactionTestCase，因为它允许真实的事务与线程交互：

import threading

from django.contrib.auth import get_user_model
from django.test import TransactionTestCase

from credits.models import CreditAccount, Generation
from credits.services import request_generation_buggy


class RaceConditionTest(TransactionTestCase):
    def test_two_requests_spend_the_same_credit(self):
        user = get_user_model().objects.create_user("demo", password="demo")
        account = CreditAccount.objects.create(user=user, balance=1)

        results = []
        barrier = threading.Barrier(2)

        def worker():
            barrier.wait()
            results.append(request_generation_buggy(account.id, "a cat"))

        threads = [threading.Thread(target=worker) for _ in range(2)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        accepted = [r for r in results if r is not None]
        account.refresh_from_db()
        self.assertEqual(account.balance, 0)
        self.assertEqual(len(accepted), 2)
        self.assertEqual(Generation.objects.count(), 2)

threading.Barrier(2) 让两个线程在真正发起请求前对齐，最大化重叠的机会。运行这个测试，你会看到余额是 0，但被接受的请求有两条、生成记录也有两条——应用接受了它付不起的工作。这就是竞态条件。

第七步：用事务与行锁修复

修复思路是：把「读余额、判断、扣减、写记录」放进同一个数据库事务，并在读取账号行时加行级锁，让第二个请求必须等第一个事务提交后才能读到余额。Django 提供了 transaction.atomic() 和 select_for_update()：

from django.db import transaction

from .models import CreditAccount, Generation


def request_generation(account_id, prompt):
    with transaction.atomic():
        account = (
            CreditAccount.objects.select_for_update()
            .get(id=account_id)
        )
        if account.balance < 1:
            return None
        account.balance -= 1
        account.save(update_fields=["balance"])
        return Generation.objects.create(account=account, prompt=prompt)

select_for_update() 会生成 SELECT ... FOR UPDATE，在事务提交或回滚前锁住这一行。第二个请求执行到同一句时会被阻塞，等第一个事务提交后再读取，此时它看到的是已经扣减过的余额，于是正确地返回 None。余额判断和扣减都在锁的保护范围内，中间不再有可被插入的窗口。

把测试里的函数换成 request_generation，断言相应改为：余额为 0、被接受的请求恰好 1 条、生成记录恰好 1 条。这就是修复后的预期行为。

第八步：接上接口

在 credits/views.py 中写一个接收 prompt 的接口，调用修复后的服务函数：

from rest_framework import status
from rest_framework.response import Response
from rest_framework.views import APIView

from .models import CreditAccount
from .services import request_generation


class GenerateView(APIView):
    def post(self, request):
        prompt = request.data.get("prompt")
        if not prompt:
            return Response(
                {"detail": "prompt is required"},
                status=status.HTTP_400_BAD_REQUEST,
            )

        account, _ = CreditAccount.objects.get_or_create(user=request.user)
        generation = request_generation(account.id, prompt)
        if generation is None:
            return Response(
                {"detail": "insufficient credits"},
                status=status.HTTP_402_PAYMENT_REQUIRED,
            )
        return Response(
            {"id": generation.id, "prompt": generation.prompt},
            status=status.HTTP_201_CREATED,
        )

在 config/urls.py 中挂上路由：

from django.urls import path

from credits.views import GenerateView

urlpatterns = [
    path("api/generate/", GenerateView.as_view()),
]

接口需要认证，用 Basic 认证带上用户名和密码即可。额度不足时返回 402，成功时返回 201 和生成记录。

一个完整示例

把上面的步骤串起来，从零跑通一次：

1. 按第一步创建目录、虚拟环境，写好 requirements.txt 并安装依赖，创建 config 项目和 credits 应用。
2. 按第二步写好 compose.yaml，执行 docker compose up -d --wait，确认容器健康。
3. 按第三步替换 config/settings.py，按第四步写好模型并执行 makemigrations 与 migrate。
4. 先写第五步的有 bug 版本，跑第六步的测试，观察余额为 0 却有两条生成记录。
5. 换成第七步的 request_generation，再跑一次测试，确认只接受一条请求。
6. 按第八步接上接口与路由，启动开发服务器：

python manage.py runserver

先创建一个用户并给它 1 分额度，然后用两个终端几乎同时发请求：

curl -u demo:demo -X POST http://127.0.0.1:8000/api/generate/ \
  -H "Content-Type: application/json" \
  -d '{"prompt": "a cat"}'

修复后的版本里，两个请求只有一个会拿到 201，另一个拿到 402；数据库里余额为 0，生成记录只有一条。把额度改成 10 再试，可以观察到并发请求被逐个串行化，余额始终不会变成负数。

注意事项

• 数据库必须支持行锁。SQLite 没有实现 select_for_update() 依赖的行锁，所以示例用 PostgreSQL。换数据库前先确认它是否支持。
• 锁只在事务内有效。select_for_update() 必须在 transaction.atomic() 块里调用，否则锁会立刻释放，等于没加。
• 超时是错误路径。示例设置了 5 秒锁超时和 10 秒语句超时，锁等待超时会直接抛错。这个精简接口没有为超时提供自定义响应，生产环境需要自行处理。
• 行锁不是万能解。前面提到的旧搜索响应覆盖新结果，属于客户端时序问题，应该用请求标识或过期响应检查，而不是数据库锁。
• 唯一性约束要落在数据库上。用户名重复这类问题，不能只靠前置检查，必须在数据库层面强制唯一。
• 示例配置仅供本地。示例中的密钥、DEBUG = True、明文密码和回环地址端口映射都不适合生产；部署时请配置生产密钥、合适的认证方式与加密连接。
• 版本与配额以官网为准。文中涉及的 Django、DRF、PostgreSQL 版本号以及任何额度规则都可能变化，实际使用前请查阅各项目官网的当前信息。