Django后端并发请求处理：保证StockPurchase数据一致性方案

背景 #

我有两个API端点：

• stockpurchases/ || stockpurchases/{id}/
• stockpurchases/{id}/stocks/ || stockpurchases/{id}/stocks/{id}/

业务逻辑：用户需先创建StockPurchase，再向其中添加Stock；每次POST创建Stock时，需累加对应StockPurchase的qty和amount字段值。

当前问题：循环连续发送无等待的POST请求时，StockPurchase的amount计算会出现错误。

需求：让Django后端处理多请求时，后续请求等待前一个请求处理完成后再执行，保证数据计算正确。

相关模型代码 #

StockPurchase模型 #

class StockPurchase(models.Model):
    qty = models.IntegerField(default=0)
    amount = models.IntegerField(default=0)

Stock模型 #

class Stock(models.Model):
    stockpurchase = models.ForeignKey("StockPurchase", on_delete=models.CASCADE, related_name='stocks')
    product = models.ForeignKey("Product", on_delete=models.CASCADE, related_name='stocks')
    project = models.ForeignKey("Project", on_delete=models.CASCADE, related_name='stocks')

    rate = models.IntegerField()

    def save(self, *args, **kwargs):
        if not self.pk:
            self.stockpurchase.amount += self.rate
            self.stockpurchase.qty += 1
            self.stockpurchase.save()
        super().save()

测试场景 #

• 计算正确的情况：使用带await的异步循环调用，请求串行执行，数据计算无误：

async function create() {
  for (let i = 0; i < 10; i++) {
    let response = await fetch(
      "http://localhost:8000/api/stockpurchases/2/stocks/",
      {
        method: "POST",
        body: bodyContent,
        headers: headersList,
      }
    );
  }
}

• 计算错误的情况：使用不带await的循环调用，请求并发发送，StockPurchase的amount计算结果不符合预期：

for (let i = 0; i < 10; i++) {
  let response = fetch("http://localhost:8000/api/stockpurchases/2/stocks/", {
    method: "POST",
    body: bodyContent,
    headers: headersList,
  });
}

问题原因与解决方案 #

原因分析 #

问题源于并发竞争条件：多个请求同时读取同一个StockPurchase对象的初始值，各自修改后保存，导致后续请求覆盖了前面的修改结果。比如两个请求同时读取amount=10，各自加5后都存为15，而非正确的20。

解决方法 #

方案1：使用行级锁+事务控制

在Stock创建视图中，通过select_for_update()获取StockPurchase对象并加行级锁，确保同一时间只有一个请求能修改该对象：

from django.db import transaction
from rest_framework import generics

class StockCreateView(generics.CreateAPIView):
    queryset = Stock.objects.all()
    serializer_class = StockSerializer

    def perform_create(self, serializer):
        stock_purchase_id = self.kwargs['id']
        # 开启事务并加行级锁
        with transaction.atomic():
            stock_purchase = StockPurchase.objects.select_for_update().get(id=stock_purchase_id)
            stock = serializer.save(stockpurchase=stock_purchase)
            # 累加字段
            stock_purchase.amount += stock.rate
            stock_purchase.qty += 1
            stock_purchase.save()

同时修改Stock模型的save方法，移除原有的累加逻辑，避免重复操作：

class Stock(models.Model):
    # 字段定义不变
    def save(self, *args, **kwargs):
        super().save(*args, **kwargs)

方案2：使用F表达式进行原子更新

通过Django的F()表达式直接在数据库层面完成原子更新，跳过读取-修改-写入的环节，从根源避免竞争：

from django.db.models import F
from rest_framework import generics

class StockCreateView(generics.CreateAPIView):
    queryset = Stock.objects.all()
    serializer_class = StockSerializer

    def perform_create(self, serializer):
        stock = serializer.save()
        # 数据库层面原子更新字段
        StockPurchase.objects.filter(id=stock.stockpurchase_id).update(
            amount=F('amount') + stock.rate,
            qty=F('qty') + 1
        )

这种方式无需额外事务控制，因为update操作是数据库原生的原子操作，能保证多请求同时更新时不会出现数据覆盖。

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