基于模型计数限制奖品兑换时的竞态条件解决方案咨询

针对并发场景下奖品超额兑换的问题，结合你的需求（避免锁错误、正确拒绝或排队），以下是几种落地性强的解决方案：

方案1：乐观锁+原子更新（推荐） #

通过在Prize模型新增计数字段，利用Django的F()表达式实现原子化的计数检查与更新，锁持有时间极短，适合高并发场景。

步骤1：修改Prize模型 #

新增redeemed_count字段记录已兑换次数：

from django.db import models

class Prize(models.Model):
    redemption_limit = models.IntegerField(verbose_name="兑换上限")
    redeemed_count = models.IntegerField(verbose_name="已兑换次数", default=0)
    # 其他原有字段

步骤2：视图实现 #

from django.db import transaction
from django.db.models import F

def redeem(request, prize_id):
    try:
        with transaction.atomic():
            # 原子化检查并更新计数：仅当已兑换次数小于上限时，才递增计数
            updated_rows = Prize.objects.filter(
                id=prize_id,
                redeemed_count__lt=F('redemption_limit')
            ).update(
                redeemed_count=F('redeemed_count') + 1
            )
            
            if updated_rows == 0:
                # 无行被更新，说明已达兑换上限
                return error_page("Reached redemption limit")
            
            # 在原子块内创建兑换记录，保证计数与记录一致性
            Redemption.objects.create(prize_id=prize_id, status="pending")
        
        # 外部API调用放在事务外，避免长时间持有数据库锁
        api_success = run_api_redeem(prize_id)
        if api_success:
            Redemption.objects.filter(prize_id=prize_id).last().update(status="success")
            return success_page("Redeemed successfully")
        else:
            # API调用失败，回滚计数并标记状态
            with transaction.atomic():
                Prize.objects.filter(id=prize_id).update(redeemed_count=F('redeemed_count') - 1)
                Redemption.objects.filter(prize_id=prize_id).last().update(status="failed")
            return error_page("Redemption failed, please try again")
    
    except Exception:
        return error_page("Please try again later")

优缺点 #

• 优点：锁持有时间短，高并发下性能优异；Django ORM原生支持，代码易维护；天然避免竞态条件。
• 缺点：需新增字段，若已有大量历史数据需做初始化同步；API失败时需额外处理回滚逻辑。

方案2：select_for_update+原子事务（易理解） #

利用数据库行锁，保证同一时间只有一个请求能检查兑换计数并创建记录，适合并发量中等、逻辑简单的场景。

视图实现 #

from django.db import transaction, OperationalError

def redeem(request, prize_id):
    try:
        with transaction.atomic():
            # 对Prize行加排他锁，阻塞其他请求修改该奖品的相关数据
            prize = Prize.objects.select_for_update().get(id=prize_id)
            
            # 此时计数查询是绝对准确的（无其他请求能插入Redemption）
            if prize.redemption_set.count() >= prize.redemption_limit:
                return error_page("Reached redemption limit")
            
            # 先创建兑换记录，再调用API（或调整顺序，根据API幂等性决定）
            redemption = Redemption.objects.create(prize=prize)
        
        api_success = run_api_redeem(prize)
        if api_success:
            redemption.status = "success"
            redemption.save()
            return success_page("Redeemed successfully")
        else:
            redemption.status = "failed"
            redemption.save()
            return error_page("Redemption failed, please try again")
    
    except OperationalError:
        # 处理锁等待超时（MySQL默认锁等待超时50秒，可通过innodb_lock_wait_timeout调整）
        return error_page("Please try again later")
    except Prize.DoesNotExist:
        return error_page("Prize not found")

优缺点 #

• 优点：无需修改模型，逻辑直观；事务内操作完全原子，无数据一致性问题。
• 缺点：锁持有时间较长（包含API调用的话更久，所以建议API放在事务外），高并发下会出现请求排队，用户可能需要重试。

方案3：原生SQL原子插入（数据库层面保障） #

通过MySQL的INSERT ... SELECT语句，在数据库层面完成“检查上限+插入记录”的原子操作，无需Django层面处理锁逻辑。

视图实现 #

from django.db import connection, IntegrityError

def redeem(request, prize_id):
    try:
        with connection.cursor() as cursor:
            # 原子执行：仅当奖品未达兑换上限时，插入Redemption记录
            cursor.execute("""
                INSERT INTO redemption (prize_id, status)
                SELECT %s, 'pending'
                FROM prize
                WHERE id = %s AND redemption_limit > (
                    SELECT COUNT(*) FROM redemption WHERE prize_id = %s
                )
            """, [prize_id, prize_id, prize_id])
            
            if cursor.rowcount == 0:
                # 无记录插入，说明已达上限
                return error_page("Reached redemption limit")
        
        # 调用外部API并更新状态
        api_success = run_api_redeem(prize_id)
        if api_success:
            cursor.execute("""
                UPDATE redemption SET status = 'success' WHERE prize_id = %s ORDER BY id DESC LIMIT 1
            """, [prize_id])
            return success_page("Redeemed successfully")
        else:
            cursor.execute("""
                UPDATE redemption SET status = 'failed' WHERE prize_id = %s ORDER BY id DESC LIMIT 1
            """, [prize_id])
            return error_page("Redemption failed, please try again")
    
    except IntegrityError:
        return error_page("Please try again later")

优缺点 #

• 优点：完全依赖数据库原子性，无需Django层面锁逻辑；性能优异。
• 缺点：需要写原生SQL，对数据库语法依赖度高；调试和维护成本略高。

额外注意事项 #

1. API幂等性：外部兑换API必须保证幂等性（重复调用不会重复发奖），避免API重试导致的超额发放。
2. 异常处理：所有方案都需处理数据库异常（如锁超时、连接失败），返回友好的重试提示，而非直接抛出错误。
3. 数据一致性：若Redemption记录需作为权威数据，可通过MySQL触发器同步Prize.redeemed_count与redemption表的计数，避免手动更新导致的不一致。

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