如何利用Django @transaction.atomic()防止并发创建Ticket模型对象？

这个问题我之前做票务系统时也碰到过——并发请求下，两个用户同时抢同一个座位，因为都先查询到座位未被占用，接着都执行创建操作，最后触发数据库的unique_together约束报错。用@transaction.atomic()结合适当的处理逻辑，就能完美解决这个问题，下面给你两种实用的方案：

方案一：利用事务原子性 + 捕获唯一约束异常 #

这种方式直接依赖数据库的unique_together约束，在事务中尝试创建Ticket，捕获并发导致的IntegrityError，逻辑简洁高效。

首先导入需要的模块：

from django.db import transaction, IntegrityError
from .models import Ticket

然后编写创建Ticket的原子函数：

@transaction.atomic()
def create_ticket(show_id, seat_id, user_id):
    try:
        # 在原子事务中尝试创建Ticket
        ticket = Ticket.objects.create(
            show_id=show_id,
            seat_id=seat_id,
            user_id=user_id
        )
        return ticket
    except IntegrityError:
        # 捕获唯一约束异常，说明该座位已被其他请求抢占
        return None

逻辑说明： #

• @transaction.atomic()保证整个创建操作是原子性的，要么全部成功，要么全部回滚。
• 当第一个请求成功创建Ticket后，第二个并发请求执行create()时，数据库会因为unique_together约束抛出IntegrityError，我们捕获这个异常就可以判断座位已被占用，返回None做后续处理。

方案二：事务内加行锁提前避免冲突 #

如果需要在创建前做更多业务检查（比如判断演出是否已开始），可以用select_for_update()对关联的Show和Seat行加排他锁，阻塞其他并发事务的操作，从根源上避免冲突。

代码示例：

from django.db import transaction
from django.utils import timezone
from .models import Ticket, Show, Seat

@transaction.atomic()
def create_ticket(show_id, seat_id, user_id):
    # 对Show和Seat行加排他锁，直到事务结束才释放
    show = Show.objects.select_for_update().get(id=show_id)
    seat = Seat.objects.select_for_update().get(id=seat_id)
    
    # 这里可以增加额外的业务检查，比如判断演出是否已开始
    if show.start_time <= timezone.now():
        return None, "演出已开始，无法预订座位"
    
    # 检查该座位是否已被预订
    if Ticket.objects.filter(show=show, seat=seat).exists():
        return None, "该座位已被预订"
    
    # 创建Ticket
    ticket = Ticket.objects.create(
        show=show,
        seat=seat,
        user_id=user_id
    )
    return ticket, None

逻辑说明： #

• select_for_update()会在查询Show和Seat时加排他锁，其他并发事务尝试查询同一行时会被阻塞，直到当前事务提交或回滚。
• 这样就能保证同一时间只有一个事务能执行后续的检查和创建操作，彻底避免并发冲突。

在视图中使用示例（以DRF为例） #

把上面的函数集成到视图里，给用户返回友好的响应：

from rest_framework.views import APIView
from rest_framework.response import Response
from .serializers import TicketSerializer

class TicketCreateView(APIView):
    def post(self, request):
        show_id = request.data.get('show_id')
        seat_id = request.data.get('seat_id')
        user_id = request.user.id
        
        ticket, error = create_ticket(show_id, seat_id, user_id)
        if ticket:
            serializer = TicketSerializer(ticket)
            return Response(serializer.data, status=201)
        else:
            return Response({"error": error}, status=400)

两种方案各有优劣：方案一代码更简洁，适合简单场景；方案二灵活性更高，适合需要额外业务校验的场景，你可以根据自己的需求选择。

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