Django 4.1中按顺序分配数据集避免竞态条件的实现方案

竞态风险确实存在 #

当数百名用户同时请求任务时，如果没有额外的并发控制，多个请求会同时查询到同一批未分配的任务，随后各自将其标记为已领取，最终导致重复分配。这是典型的竞态条件问题——数据库的查询与更新是两个独立操作，中间存在时间窗口，让其他请求有机可乘。

解决方案 #

行级锁 + 原子事务 #

这是最直接可靠的方案，利用数据库行级锁机制，在查询任务时锁定目标行，确保同一时间只有一个事务能修改它。结合Django的atomic装饰器，保证查询与更新操作的原子性，避免中间状态暴露给其他请求。

示例代码：

from django.db import transaction
from .models import Task

@transaction.atomic
def assign_next_task(user):
    # 锁定未分配的第一条任务，其他请求会等待锁释放后再执行
    task = Task.objects.select_for_update().filter(status='pending').order_by('id').first()
    if task:
        task.status = 'assigned'
        task.assigned_to = user
        task.save()
        return task
    return None

注意：select_for_update会阻塞其他请求直到锁释放，高并发场景下可能产生等待队列，需要根据业务的延迟容忍度评估使用。

乐观锁 #

适合并发量高但任务冲突概率较低的场景，无需显式加锁，通过版本号或时间戳验证更新时数据是否被篡改。

首先给Task模型添加版本字段：

class Task(models.Model):
    # 原有字段...
    status = models.CharField(max_length=20, default='pending')
    version = models.IntegerField(default=0)

分配任务的逻辑：

def assign_next_task(user):
    while True:
        # 查询未分配的第一条任务
        task = Task.objects.filter(status='pending').order_by('id').first()
        if not task:
            return None
        
        # 更新时验证版本号，只有版本一致才会执行更新
        updated_count = Task.objects.filter(id=task.id, version=task.version).update(
            status='assigned',
            assigned_to=user,
            version=task.version + 1
        )
        
        if updated_count == 1:
            # 更新成功，刷新任务数据后返回
            task.refresh_from_db()
            return task
        # 更新失败，说明任务已被其他用户领取，重新循环尝试

这种方式不会阻塞请求，冲突时自动重试，对数据库压力更小，但冲突频繁时会增加重试次数，需注意设置合理的重试上限。

任务队列分发 #

如果任务分配逻辑固定，可以将未分配任务提前存入线程安全的队列（如Redis队列），用户请求时直接从队列头部取出任务。队列本身保证了同一时间只有一个消费者能获取任务，从根源上避免数据库层面的竞态。

大致实现思路：

1. 系统初始化或定时任务将所有pending任务批量写入Redis队列
2. 用户请求任务时，从Redis队列弹出头部元素
3. 根据弹出的任务ID，更新数据库中对应任务的状态

这种方案适合超高并发场景，数据库仅负责持久化状态，分配逻辑由队列处理，性能优势明显。

选型建议 #

• 中小并发场景：优先选择行级锁+原子事务，实现简单、可靠性高
• 高并发低冲突场景：使用乐观锁，兼顾性能与并发能力
• 超高并发场景：采用任务队列，彻底隔离分配逻辑与数据库操作

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