CodeWalk

Django ORM事务隔离级别与分布式锁的配合

作者：苦行僧 · 2026-05-30 12:55

Django ORM提供了transaction.atomic()上下文管理器进行事务控制。请介绍Django ORM中如何设置事务隔离级别（transaction.set_isolation_level()/数据库配置）、select_for_update()行级锁的使用，以及在高并发场景下如何配合Redis分布式锁防止超卖/重复提交。

回答

苦行僧

Django事务基础

from django.db import transaction

# 原子事务
with transaction.atomic():
    user = User.objects.select_for_update().get(pk=1)
    user.balance -= 100
    user.save()
    Order.objects.create(user=user, amount=100)
# 自动commit或rollback

# 手动控制
try:
    with transaction.atomic():
        # 数据库操作
        raise SomeError()
        # 到这里会自动rollback
except:
    pass  # 事务已回滚

事务隔离级别设置

# 方式1：数据库配置（settings.py）
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'OPTIONS': {
            'isolation_level': 'repeatable_read',
        },
    }
}

# 方式2：session级别（Django 3.0+）
from django.db import connection

with connection.cursor() as cursor:
    cursor.execute('SET TRANSACTION ISOLATION LEVEL SERIALIZABLE')

select_for_update() — 行级锁

# 悲观锁：锁定行直到事务结束
with transaction.atomic():
    # 添加nowait=True防止等待锁（直接报错）
    # 添加skip_locked=True跳过已被锁的行
    user = User.objects.select_for_update(nowait=True).get(pk=1)
    if user.balance >= 100:
        user.balance -= 100
        user.save()

分布式锁+事务（防超卖）

import redis
from django.db import transaction

redis_client = redis.Redis()

lock_key = f'product_lock:{product_id}'
# 获取Redis分布式锁
with redis_client.lock(lock_key, timeout=10):
    with transaction.atomic():
        product = Product.objects.select_for_update().get(pk=product_id)
        if product.stock <= 0:
            raise Exception('库存不足')
        product.stock -= 1
        product.save()
        Order.objects.create(
            user=request.user,
            product=product,
            quantity=1
        )

最佳实践：

  1. 单数据库：select_for_update() + transaction.atomic()
  2. 跨数据库/微服务：Redis分布式锁 + 重试机制
  3. 高并发秒杀：Redis原子操作预扣 + 异步订单创建